Repository: matrixorigin/matrixone
Pull request #3105 (2022-06-19 07:46:54)

fix a bug in any_value
**What type of PR is this?**

- [ ] API-change
- [x] BUG
- [ ] Improvement
- [ ] Documentation
- [ ] Feature
- [ ] Test and CI
- [ ] Code Refactoring

**Which issue(s) this PR fixes:**

issue #3051 

**What this PR does / why we need it:**

Not Available

**Special notes for your reviewer:**

Not Available

**Additional documentation (e.g. design docs, usage docs, etc.):**

Not Available



Patch:
diff --git a/pkg/container/ring/anyvalue/anyvalue.go b/pkg/container/ring/anyvalue/anyvalue.go
index d8a00ad27..006d0918a 100644
--- a/pkg/container/ring/anyvalue/anyvalue.go
+++ b/pkg/container/ring/anyvalue/anyvalue.go
@@ -194,6 +194,10 @@ func NewAnyValueRingWithTypeCheck(typ types.Type) (ring.Ring, error) {
 		return &AnyVRing1[int32]{Typ: typ}, nil
 	case types.T_int64:
 		return &AnyVRing1[int64]{Typ: typ}, nil
+	case types.T_float32:
+		return &AnyVRing1[float32]{Typ: typ}, nil
+	case types.T_float64:
+		return &AnyVRing1[float64]{Typ: typ}, nil
 	case types.T_char:
 		return &AnyVRing2{Typ: typ}, nil
 	case types.T_varchar:
diff --git a/pkg/hakeeper/checker.go b/pkg/hakeeper/checker.go
index 3c4801e90..e91591f28 100644
--- a/pkg/hakeeper/checker.go
+++ b/pkg/hakeeper/checker.go
@@ -35,6 +35,6 @@ type Checker interface {
 	// Check is periodically called by the HAKeeper for checking the cluster
 	// health status, a list of Operator instances will be returned describing
 	// actions required to ensure the high availability of the cluster.
-	Check(alloc IDAllocator, cluster ClusterInfo,
-		dn DNState, log LogState, currentTick uint64) []hapb.ScheduleCommand
+	Check(alloc IDAllocator, cluster hapb.ClusterInfo,
+		dn hapb.DNState, log hapb.LogState, currentTick uint64) []hapb.ScheduleCommand
 }
diff --git a/pkg/hakeeper/checkers/dnservice/check.go b/pkg/hakeeper/checkers/dnservice/check.go
index d7695711b..db94cd059 100644
--- a/pkg/hakeeper/checkers/dnservice/check.go
+++ b/pkg/hakeeper/checkers/dnservice/check.go
@@ -18,7 +18,7 @@ import (
 	"fmt"
 	"sort"
 
-	"github.com/matrixorigin/matrixone/pkg/hakeeper"
+	"github.com/matrixorigin/matrixone/pkg/pb/hakeeper"
 )
 
 // FIXME: placeholder
diff --git a/pkg/hakeeper/checkers/dnservice/parse.go b/pkg/hakeeper/checkers/dnservice/parse.go
index eb7568622..b495bc027 100644
--- a/pkg/hakeeper/checkers/dnservice/parse.go
+++ b/pkg/hakeeper/checkers/dnservice/parse.go
@@ -18,6 +18,7 @@ import (
 	"time"
 
 	"github.com/matrixorigin/matrixone/pkg/hakeeper"
+	hapb "github.com/matrixorigin/matrixone/pkg/pb/hakeeper"
 )
 
 const (
@@ -27,7 +28,7 @@ const (
 
 // parseDnState parses cluster dn state.
 func parseDnState(
-	dnState hakeeper.DNState, currTick uint64,
+	dnState hapb.DNState, currTick uint64,
 ) (*clusterStores, *clusterShards) {
 	stores := newClusterStores()
 	shards := newClusterShards()
diff --git a/pkg/hakeeper/checkers/logservice/checker.go b/pkg/hakeeper/checkers/logservice/checker.go
index f1408dcd8..301cf8bc6 100644
--- a/pkg/hakeeper/checkers/logservice/checker.go
+++ b/pkg/hakeeper/checkers/logservice/checker.go
@@ -16,9 +16,9 @@ package logservice
 import (
 	"sort"
 
-	"github.com/matrixorigin/matrixone/pkg/hakeeper"
 	"github.com/matrixorigin/matrixone/pkg/hakeeper/operator"
 	"github.com/matrixorigin/matrixone/pkg/hakeeper/utils"
+	"github.com/matrixorigin/matrixone/pkg/pb/hakeeper"
 	"github.com/matrixorigin/matrixone/pkg/pb/metadata"
 )
 
@@ -74,8 +74,8 @@ func collectStats(cluster hakeeper.ClusterInfo, infos hakeeper.LogState, tick ui
 		for replicaID, uuid := range shardInfo.Replicas {
 			// Check dangling
 			started := false
-			for _, shard := range infos.Stores[uuid].Shards {
-				if shard.ShardID == shardInfo.ShardID {
+			for _, replica := range infos.Stores[uuid].Replicas {
+				if replica.ShardID == shardInfo.ShardID {
 					started = true
 				}
 			}
@@ -95,9 +95,9 @@ func collectStats(cluster hakeeper.ClusterInfo, infos hakeeper.LogState, tick ui
 
 	// Check zombies
 	for uuid, storeInfo := range infos.Stores {
-		for _, shardInfo := range storeInfo.Shards {
-			if shardInfo.Epoch < infos.Shards[shardInfo.ShardID].Epoch {
-				s.toStop = append(s.toStop, replica{uuid: uuid, shardID: shardInfo.ShardID})
+		for _, replicaInfo := range storeInfo.Replicas {
+			if replicaInfo.Epoch < infos.Shards[replicaInfo.ShardID].Epoch {
+				s.toStop = append(s.toStop, replica{uuid: uuid, shardID: replicaInfo.ShardID})
 			}
 		}
 	}
diff --git a/pkg/hakeeper/operator/controller.go b/pkg/hakeeper/operator/controller.go
index 16ddeb6ff..70cf5d08e 100644
--- a/pkg/hakeeper/operator/controller.go
+++ b/pkg/hakeeper/operator/controller.go
@@ -21,7 +21,7 @@
 package operator
 
 import (
-	"github.com/matrixorigin/matrixone/pkg/hakeeper"
+	"github.com/matrixorigin/matrixone/pkg/pb/hakeeper"
 	"sync"
 )
 
diff --git a/pkg/hakeeper/operator/operator.go b/pkg/hakeeper/operator/operator.go
index 02db0dc85..81f19df7a 100644
--- a/pkg/hakeeper/operator/operator.go
+++ b/pkg/hakeeper/operator/operator.go
@@ -24,7 +24,7 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/matrixorigin/matrixone/pkg/hakeeper"
+	"github.com/matrixorigin/matrixone/pkg/pb/hakeeper"
 )
 
 const (
diff --git a/pkg/hakeeper/operator/steps.go b/pkg/hakeeper/operator/steps.go
index b00ac0185..557f7e6a2 100644
--- a/pkg/hakeeper/operator/steps.go
+++ b/pkg/hakeeper/operator/steps.go
@@ -22,7 +22,8 @@ package operator
 
 import (
 	"fmt"
-	"github.com/matrixorigin/matrixone/pkg/hakeeper"
+
+	"github.com/matrixorigin/matrixone/pkg/pb/hakeeper"
 )
 
 type OpStep interface {
@@ -77,8 +78,10 @@ func (a StartLogService) String() string {
 }
 
 func (a StartLogService) IsFinish(state hakeeper.LogState, _ hakeeper.DNState) bool {
-	for _, shardInfo := range state.Stores[a.UUID].Shards {
-		if shardInfo.ShardID == a.ShardID {
+	// FIXME: state.Stores[a.UUID] is going to return nil when a.UUID is not a
+	// key in state.Stores
+	for _, replicaInfo := range state.Stores[a.UUID].Replicas {
+		if replicaInfo.ShardID == a.ShardID {
 			return true
 		}
 	}
@@ -96,8 +99,10 @@ func (a StopLogService) String() string {
 }
 
 func (a StopLogService) IsFinish(state hakeeper.LogState, _ hakeeper.DNState) bool {
-	for _, shardInfo := range state.Stores[a.UUID].Shards {
-		if shardInfo.ShardID == a.ShardID {
+	// FIXME: state.Stores[a.UUID] is going to return nil when a.UUID is not a
+	// key in state.Stores
+	for _, replicaInfo := range state.Stores[a.UUID].Replicas {
+		if replicaInfo.ShardID == a.ShardID {
 			return false
 		}
 	}
diff --git a/pkg/hakeeper/rsm.go b/pkg/hakeeper/rsm.go
index 4268043e2..aa9757fd2 100644
--- a/pkg/hakeeper/rsm.go
+++ b/pkg/hakeeper/rsm.go
@@ -57,27 +57,12 @@ const (
 )
 
 type StateQuery struct{}
-
-type logShardIDQuery struct {
-	name string
-}
-
-type logShardIDQueryResult struct {
-	id    uint64
-	found bool
-}
+type logShardIDQuery struct{ name string }
+type logShardIDQueryResult struct{ id uint64 }
 
 type stateMachine struct {
-	replicaID        uint64
-	term             uint64
-	scheduleCommands map[string][]hapb.ScheduleCommand // keyed by UUID
-
-	Tick        uint64
-	NextID      uint64
-	LogShards   map[string]uint64 // keyed by Log Shard name
-	DNState     DNState
-	LogState    LogState
-	ClusterInfo ClusterInfo
+	replicaID uint64
+	state     hapb.RSMState
 }
 
 func parseCmdTag(cmd []byte) uint16 {
@@ -190,11 +175,8 @@ func NewStateMachine(shardID uint64, replicaID uint64) sm.IStateMachine {
 		panic(moerr.NewError(moerr.INVALID_INPUT, "invalid HAKeeper shard ID"))
 	}
 	return &stateMachine{
-		replicaID:        replicaID,
-		scheduleCommands: make(map[string][]hapb.ScheduleCommand),
-		LogShards:        make(map[string]uint64),
-		DNState:          NewDNState(),
-		LogState:         NewLogState(),
+		replicaID: replicaID,
+		state:     hapb.NewRSMState(),
 	}
 }
 
@@ -203,8 +185,8 @@ func (s *stateMachine) Close() error {
 }
 
 func (s *stateMachine) assignID() uint64 {
-	s.NextID++
-	return s.NextID
+	s.state.NextID++
+	return s.state.NextID
 }
 
 func (s *stateMachine) handleUpdateCommandsCmd(cmd []byte) sm.Result {
@@ -213,20 +195,22 @@ func (s *stateMachine) handleUpdateCommandsCmd(cmd []byte) sm.Result {
 	if err := b.Unmarshal(data); err != nil {
 		panic(err)
 	}
-	plog.Infof("incoming term: %d, rsm term: %d", b.Term, s.term)
-	if s.term > b.Term {
+	plog.Infof("incoming term: %d, rsm term: %d", b.Term, s.state.Term)
+	if s.state.Term > b.Term {
 		return sm.Result{}
 	}
 
-	s.term = b.Term
-	s.scheduleCommands = make(map[string][]hapb.ScheduleCommand)
+	s.state.Term = b.Term
+	s.state.ScheduleCommands = make(map[string]hapb.CommandBatch)
 	for _, c := range b.Commands {
-		l, ok := s.scheduleCommands[c.UUID]
+		l, ok := s.state.ScheduleCommands[c.UUID]
 		if !ok {
-			l = make([]hapb.ScheduleCommand, 0)
+			l = hapb.CommandBatch{
+				Commands: make([]hapb.ScheduleCommand, 0),
+			}
 		}
-		l = append(l, c)
-		s.scheduleCommands[c.UUID] = l
+		l.Commands = append(l.Commands, c)
+		s.state.ScheduleCommands[c.UUID] = l
 	}
 
 	return sm.Result{}
@@ -237,13 +221,13 @@ func (s *stateMachine) handleCreateLogShardCmd(cmd []byte) sm.Result {
 	if !ok {
 		panic(moerr.NewError(moerr.INVALID_INPUT, "not create log shard cmd"))
 	}
-	if shardID, ok := s.LogShards[name]; ok {
+	if shardID, ok := s.state.LogShards[name]; ok {
 		data := make([]byte, 8)
 		binaryEnc.PutUint64(data, shardID)
 		return sm.Result{Value: 0, Data: data}
 	}
-	s.LogShards[name] = s.assignID()
-	return sm.Result{Value: s.NextID}
+	s.state.LogShards[name] = s.assignID()
+	return sm.Result{Value: s.state.NextID}
 }
 
 func (s *stateMachine) handleDNHeartbeat(cmd []byte) sm.Result {
@@ -252,7 +236,7 @@ func (s *stateMachine) handleDNHeartbeat(cmd []byte) sm.Result {
 	if err := hb.Unmarshal(data); err != nil {
 		panic(err)
 	}
-	s.DNState.Update(hb, s.Tick)
+	s.state.DNState.Update(hb, s.state.Tick)
 	return sm.Result{}
 }
 
@@ -262,20 +246,20 @@ func (s *stateMachine) handleLogHeartbeat(cmd []byte) sm.Result {
 	if err := hb.Unmarshal(data); err != nil {
 		panic(err)
 	}
-	s.LogState.Update(hb, s.Tick)
+	s.state.LogState.Update(hb, s.state.Tick)
 	return sm.Result{}
 }
 
 func (s *stateMachine) handleTick(cmd []byte) sm.Result {
-	s.Tick++
+	s.state.Tick++
 	return sm.Result{}
 }
 
 func (s *stateMachine) handleGetIDCmd(cmd []byte) sm.Result {
 	count := parseGetIDCmd(cmd)
-	s.NextID++
-	v := s.NextID
-	s.NextID += (count - 1)
+	s.state.NextID++
+	v := s.state.NextID
+	s.state.NextID += (count - 1)
 	return sm.Result{Value: v}
 }
 
@@ -299,20 +283,20 @@ func (s *stateMachine) Update(e sm.Entry) (sm.Result, error) {
 
 func (s *stateMachine) handleStateQuery() interface{} {
 	// FIXME: pretty sure we need to deepcopy here
-	return &HAKeeperState{
-		Tick:        s.Tick,
-		ClusterInfo: s.ClusterInfo,
-		DNState:     s.DNState,
-		LogState:    s.LogState,
+	return &hapb.HAKeeperState{
+		Tick:        s.state.Tick,
+		ClusterInfo: s.state.ClusterInfo,
+		DNState:     s.state.DNState,
+		LogState:    s.state.LogState,
 	}
 }
 
 func (s *stateMachine) handleShardIDQuery(name string) *logShardIDQueryResult {
-	id, ok := s.LogShards[name]
+	id, ok := s.state.LogShards[name]
 	if ok {
-		return &logShardIDQueryResult{found: true, id: id}
+		return &logShardIDQueryResult{id: id}
 	}
-	return &logShardIDQueryResult{found: false}
+	return &logShardIDQueryResult{}
 }
 
 func (s *stateMachine) Lookup(query interface{}) (interface{}, error) {
@@ -326,12 +310,14 @@ func (s *stateMachine) Lookup(query interface{}) (interface{}, error) {
 
 func (s *stateMachine) SaveSnapshot(w io.Writer,
 	_ sm.ISnapshotFileCollection, _ <-chan struct{}) error {
+	// FIXME: ready to use gogoproto to marshal the state, just need to figure
+	// out how to write to the writer.
 	enc := gob.NewEncoder(w)
-	return enc.Encode(s)
+	return enc.Encode(s.state)
 }
 
 func (s *stateMachine) RecoverFromSnapshot(r io.Reader,
 	_ []sm.SnapshotFile, _ <-chan struct{}) error {
 	dec := gob.NewDecoder(r)
-	return dec.Decode(s)
+	return dec.Decode(&s.state)
 }
diff --git a/pkg/logservice/hachecker.go b/pkg/logservice/hachecker.go
index 3212addee..d992c8ccb 100644
--- a/pkg/logservice/hachecker.go
+++ b/pkg/logservice/hachecker.go
@@ -100,7 +100,7 @@ func (l *logStore) healthCheck() {
 			// TODO: check whether this is temp error
 			return
 		}
-		state := s.(*hakeeper.HAKeeperState)
+		state := s.(*hapb.HAKeeperState)
 		cmds := l.checker.Check(l.alloc,
 			state.ClusterInfo, state.DNState, state.LogState, state.Tick)
 		if len(cmds) > 0 {
diff --git a/pkg/logservice/rsm.go b/pkg/logservice/rsm.go
index bb2bc0c83..7f7ad11fb 100644
--- a/pkg/logservice/rsm.go
+++ b/pkg/logservice/rsm.go
@@ -20,6 +20,7 @@ import (
 	"io"
 
 	sm "github.com/lni/dragonboat/v4/statemachine"
+	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
 )
 
 var (
@@ -101,29 +102,29 @@ func tagMatch(cmd []byte, expectedTag uint16) bool {
 }
 
 type stateMachine struct {
-	shardID        uint64
-	replicaID      uint64
-	Index          uint64
-	LeaseHolderID  uint64
-	TruncatedIndex uint64
-	LeaseHistory   map[uint64]uint64 // log index -> truncate index
+	shardID   uint64
+	replicaID uint64
+	state     pb.RSMState
 }
 
 var _ (sm.IStateMachine) = (*stateMachine)(nil)
 
 func newStateMachine(shardID uint64, replicaID uint64) sm.IStateMachine {
-	return &stateMachine{
-		shardID:      shardID,
-		replicaID:    replicaID,
+	state := pb.RSMState{
 		LeaseHistory: make(map[uint64]uint64),
 	}
+	return &stateMachine{
+		shardID:   shardID,
+		replicaID: replicaID,
+		state:     state,
+	}
 }
 
 func (s *stateMachine) truncateLeaseHistory(index uint64) {
 	_, index = s.getLeaseHistory(index)
-	for key := range s.LeaseHistory {
+	for key := range s.state.LeaseHistory {
 		if key < index {
-			delete(s.LeaseHistory, key)
+			delete(s.state.LeaseHistory, key)
 		}
 	}
 }
@@ -131,7 +132,7 @@ func (s *stateMachine) truncateLeaseHistory(index uint64) {
 func (s *stateMachine) getLeaseHistory(index uint64) (uint64, uint64) {
 	max := uint64(0)
 	lease := uint64(0)
-	for key, val := range s.LeaseHistory {
+	for key, val := range s.state.LeaseHistory {
 		if key >= index {
 			continue
 		}
@@ -144,31 +145,31 @@ func (s *stateMachine) getLeaseHistory(index uint64) (uint64, uint64) {
 }
 
 func (s *stateMachine) handleSetLeaseHolderID(cmd []byte) sm.Result {
-	s.LeaseHolderID = parseLeaseHolderID(cmd)
-	s.LeaseHistory[s.Index] = s.LeaseHolderID
+	s.state.LeaseHolderID = parseLeaseHolderID(cmd)
+	s.state.LeaseHistory[s.state.Index] = s.state.LeaseHolderID
 	return sm.Result{}
 }
 
 func (s *stateMachine) handleTruncateIndex(cmd []byte) sm.Result {
 	index := parseTruncatedIndex(cmd)
-	if index > s.TruncatedIndex {
-		s.TruncatedIndex = index
+	if index > s.state.TruncatedIndex {
+		s.state.TruncatedIndex = index
 		s.truncateLeaseHistory(index)
 		return sm.Result{}
 	}
-	return sm.Result{Value: s.TruncatedIndex}
+	return sm.Result{Value: s.state.TruncatedIndex}
 }
 
 // handleUserUpdate returns an empty sm.Result on success or it returns a
 // sm.Result value with the Value field set to the current lease holder ID
 // to indicate rejection by mismatched lease holder ID.
 func (s *stateMachine) handleUserUpdate(cmd []byte) sm.Result {
-	if s.LeaseHolderID != parseLeaseHolderID(cmd) {
+	if s.state.LeaseHolderID != parseLeaseHolderID(cmd) {
 		data := make([]byte, 8)
-		binaryEnc.PutUint64(data, s.LeaseHolderID)
+		binaryEnc.PutUint64(data, s.state.LeaseHolderID)
 		return sm.Result{Data: data}
 	}
-	return sm.Result{Value: s.Index}
+	return sm.Result{Value: s.state.Index}
 }
 
 func (s *stateMachine) Close() error {
@@ -177,7 +178,7 @@ func (s *stateMachine) Close() error {
 
 func (s *stateMachine) Update(e sm.Entry) (sm.Result, error) {
 	cmd := e.Cmd
-	s.Index = e.Index
+	s.state.Index = e.Index
 	if isSetLeaseHolderUpdate(cmd) {
 		return s.handleSetLeaseHolderID(cmd), nil
 	} else if isSetTruncatedIndexUpdate(cmd) {
@@ -190,11 +191,11 @@ func (s *stateMachine) Update(e sm.Entry) (sm.Result, error) {
 
 func (s *stateMachine) Lookup(query interface{}) (interface{}, error) {
 	if _, ok := query.(indexQuery); ok {
-		return s.Index, nil
+		return s.state.Index, nil
 	} else if _, ok := query.(leaseHolderIDQuery); ok {
-		return s.LeaseHolderID, nil
+		return s.state.LeaseHolderID, nil
 	} else if _, ok := query.(truncatedIndexQuery); ok {
-		return s.TruncatedIndex, nil
+		return s.state.TruncatedIndex, nil
 	} else if v, ok := query.(leaseHistoryQuery); ok {
 		lease, _ := s.getLeaseHistory(v.index)
 		return lease, nil
@@ -204,12 +205,14 @@ func (s *stateMachine) Lookup(query interface{}) (interface{}, error) {
 
 func (s *stateMachine) SaveSnapshot(w io.Writer,
 	_ sm.ISnapshotFileCollection, _ <-chan struct{}) error {
+	// FIXME: use gogoproto to marshal the state, need to figure out how to
+	// marshal to a io.Writer
 	enc := gob.NewEncoder(w)
-	return enc.Encode(s)
+	return enc.Encode(s.state)
 }
 
 func (s *stateMachine) RecoverFromSnapshot(r io.Reader,
 	_ []sm.SnapshotFile, _ <-chan struct{}) error {
 	dec := gob.NewDecoder(r)
-	return dec.Decode(s)
+	return dec.Decode(&s.state)
 }
diff --git a/pkg/logservice/store.go b/pkg/logservice/store.go
index a5d7da78d..b221c556e 100644
--- a/pkg/logservice/store.go
+++ b/pkg/logservice/store.go
@@ -499,24 +499,27 @@ func (l *logStore) getHeartbeatMessage() pb.LogStoreHeartbeat {
 		RaftAddress:    l.cfg.RaftAddress,
 		ServiceAddress: l.cfg.ServiceAddress,
 		GossipAddress:  l.cfg.GossipAddress,
-		Shards:         make([]pb.LogShardInfo, 0),
+		Replicas:       make([]pb.LogReplicaInfo, 0),
 	}
 	opts := dragonboat.NodeHostInfoOption{
 		SkipLogInfo: true,
 	}
 	nhi := l.nh.GetNodeHostInfo(opts)
 	for _, ci := range nhi.ShardInfoList {
-		shardInfo := pb.LogShardInfo{
-			ShardID:  ci.ShardID,
-			Replicas: ci.Nodes,
-			Epoch:    ci.ConfigChangeIndex,
-			LeaderID: ci.LeaderID,
-			Term:     ci.Term,
+		replicaInfo := pb.LogReplicaInfo{
+			LogShardInfo: pb.LogShardInfo{
+				ShardID:  ci.ShardID,
+				Replicas: ci.Nodes,
+				Epoch:    ci.ConfigChangeIndex,
+				LeaderID: ci.LeaderID,
+				Term:     ci.Term,
+			},
+			ReplicaID: ci.ReplicaID,
 		}
-		if shardInfo.Replicas == nil {
-			shardInfo.Replicas = make(map[uint64]dragonboat.Target)
+		if replicaInfo.Replicas == nil {
+			replicaInfo.Replicas = make(map[uint64]dragonboat.Target)
 		}
-		m.Shards = append(m.Shards, shardInfo)
+		m.Replicas = append(m.Replicas, replicaInfo)
 	}
 	return m
 }
diff --git a/pkg/hakeeper/state.go b/pkg/pb/hakeeper/hakeeper.go
similarity index 58%
rename from pkg/hakeeper/state.go
rename to pkg/pb/hakeeper/hakeeper.go
index 00a3041a8..a6f22c2e1 100644
--- a/pkg/hakeeper/state.go
+++ b/pkg/pb/hakeeper/hakeeper.go
@@ -18,41 +18,30 @@ import (
 	"reflect"
 
 	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
-	metapb "github.com/matrixorigin/matrixone/pkg/pb/metadata"
+	"github.com/matrixorigin/matrixone/pkg/pb/metadata"
 )
 
-// FIXME: dragonboat should have a public value indicating what is NoLeader node id
 const (
 	// NoLeader is the replica ID of the leader node.
 	NoLeader uint64 = 0
 )
 
-type HAKeeperState struct {
-	Tick        uint64
-	ClusterInfo ClusterInfo
-	DNState     DNState
-	LogState    LogState
-}
-
-// ClusterInfo provides a global view of all shards in the cluster. It
-// describes the logical sharding of the system, rather than physical
-// distribution of all replicas that belong to those shards.
-type ClusterInfo struct {
-	DNShards  []metapb.DNShardRecord
-	LogShards []metapb.LogShardRecord
-}
-
-// DNStoreInfo contins information on a list of shards.
-type DNStoreInfo struct {
-	Tick   uint64
-	Shards []pb.DNShardInfo
+// NewRSMState creates a new RSMState instance.
+func NewRSMState() RSMState {
+	return RSMState{
+		ScheduleCommands: make(map[string]CommandBatch),
+		LogShards:        make(map[string]uint64),
+		DNState:          NewDNState(),
+		LogState:         NewLogState(),
+		ClusterInfo:      newClusterInfo(),
+	}
 }
 
-// DNState contains all DN details known to the HAKeeper.
-type DNState struct {
-	// Stores is keyed by DN store UUID, it contains details found on each DN
-	// store. Each DNStoreInfo reflects what was last reported by each DN store.
-	Stores map[string]DNStoreInfo
+func newClusterInfo() ClusterInfo {
+	return ClusterInfo{
+		DNShards:  make([]metadata.DNShardRecord, 0),
+		LogShards: make([]metadata.LogShardRecord, 0),
+	}
 }
 
 // NewDNState creates a new DNState.
@@ -75,26 +64,6 @@ func (s *DNState) Update(hb pb.DNStoreHeartbeat, tick uint64) {
 	s.Stores[hb.UUID] = storeInfo
 }
 
-// LogStoreInfo contains information of all replicas found on a Log store.
-type LogStoreInfo struct {
-	Tick           uint64
-	RaftAddress    string
-	ServiceAddress string
-	GossipAddress  string
-	Shards         []pb.LogShardInfo
-}
-
-type LogState struct {
-	// Shards is keyed by ShardID, it contains details aggregated from all Log
-	// stores. Each pb.LogShardInfo here contains data aggregated from
-	// different replicas and thus reflect a more accurate description on each
-	// shard.
-	Shards map[uint64]pb.LogShardInfo
-	// Stores is keyed by log store UUID, it contains details found on each store.
-	// Each LogStoreInfo here reflects what was last reported by each Log store.
-	Stores map[string]LogStoreInfo
-}
-
 // NewLogState creates a new LogState.
 func NewLogState() LogState {
 	return LogState{
@@ -119,12 +88,12 @@ func (s *LogState) updateStores(hb pb.LogStoreHeartbeat, tick uint64) {
 	storeInfo.RaftAddress = hb.RaftAddress
 	storeInfo.ServiceAddress = hb.ServiceAddress
 	storeInfo.GossipAddress = hb.GossipAddress
-	storeInfo.Shards = hb.Shards
+	storeInfo.Replicas = hb.Replicas
 	s.Stores[hb.UUID] = storeInfo
 }
 
 func (s *LogState) updateShards(hb pb.LogStoreHeartbeat) {
-	for _, incoming := range hb.Shards {
+	for _, incoming := range hb.Replicas {
 		recorded, ok := s.Shards[incoming.ShardID]
 		if !ok {
 			recorded = pb.LogShardInfo{
@@ -138,9 +107,7 @@ func (s *LogState) updateShards(hb pb.LogStoreHeartbeat) {
 			recorded.Replicas = incoming.Replicas
 		} else if incoming.Epoch == recorded.Epoch && incoming.Epoch > 0 {
 			if !reflect.DeepEqual(recorded.Replicas, incoming.Replicas) {
-				plog.Panicf("inconsistent replicas, %+v, %+v, nil: %t, nil: %t",
-					recorded.Replicas, incoming.Replicas,
-					recorded.Replicas == nil, incoming.Replicas == nil)
+				panic("inconsistent replicas")
 			}
 		}
 
diff --git a/pkg/pb/hakeeper/hakeeper.pb.go b/pkg/pb/hakeeper/hakeeper.pb.go
index b2d8f8fed..3fd53743d 100644
--- a/pkg/pb/hakeeper/hakeeper.pb.go
+++ b/pkg/pb/hakeeper/hakeeper.pb.go
@@ -7,6 +7,8 @@ import (
 	fmt "fmt"
 	_ "github.com/gogo/protobuf/gogoproto"
 	proto "github.com/gogo/protobuf/proto"
+	logservice "github.com/matrixorigin/matrixone/pkg/pb/logservice"
+	metadata "github.com/matrixorigin/matrixone/pkg/pb/metadata"
 	io "io"
 	math "math"
 	math_bits "math/bits"
@@ -320,323 +322,2362 @@ func (m *CommandBatch) GetCommands() []ScheduleCommand {
 	return nil
 }
 
-func init() {
-	proto.RegisterEnum("hakeeper.ConfigChangeType", ConfigChangeType_name, ConfigChangeType_value)
-	proto.RegisterEnum("hakeeper.ServiceType", ServiceType_name, ServiceType_value)
-	proto.RegisterType((*Replica)(nil), "hakeeper.Replica")
-	proto.RegisterType((*ConfigChange)(nil), "hakeeper.ConfigChange")
-	proto.RegisterType((*ScheduleCommand)(nil), "hakeeper.ScheduleCommand")
-	proto.RegisterType((*CommandBatch)(nil), "hakeeper.CommandBatch")
+// DNStoreInfo contins information on a list of shards.
+type DNStoreInfo struct {
+	Tick                 uint64                   `protobuf:"varint,1,opt,name=Tick,proto3" json:"Tick,omitempty"`
+	Shards               []logservice.DNShardInfo `protobuf:"bytes,2,rep,name=Shards,proto3" json:"Shards"`
+	XXX_NoUnkeyedLiteral struct{}                 `json:"-"`
+	XXX_unrecognized     []byte                   `json:"-"`
+	XXX_sizecache        int32                    `json:"-"`
 }
 
-func init() { proto.RegisterFile("hakeeper.proto", fileDescriptor_5e1506f3aa5330eb) }
-
-var fileDescriptor_5e1506f3aa5330eb = []byte{
-	// 400 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x6c, 0x52, 0xc1, 0xae, 0xd2, 0x40,
-	0x14, 0xed, 0x50, 0x14, 0xb8, 0x45, 0xac, 0x13, 0x35, 0x95, 0x98, 0x4a, 0xba, 0x22, 0x44, 0x21,
-	0xd6, 0x85, 0x89, 0x6e, 0x14, 0x70, 0x41, 0x42, 0x5c, 0xb4, 0x10, 0x97, 0xa6, 0xb4, 0x43, 0xdb,
-	0x48, 0x3b, 0x4d, 0x2d, 0x24, 0x2e, 0xfc, 0x17, 0x3f, 0x87, 0x25, 0x5f, 0x60, 0xde, 0xe3, 0x4b,
-	0x5e, 0x7a, 0x3b, 0x94, 0xf6, 0xe5, 0xed, 0xee, 0xe9, 0x3d, 0xe7, 0x9e, 0x73, 0x26, 0x85, 0x5e,
-	0xe0, 0xfc, 0x62, 0x2c, 0x61, 0xe9, 0x38, 0x49, 0x79, 0xc6, 0x69, 0xfb, 0x82, 0xfb, 0xef, 0xfc,
-	0x30, 0x0b, 0xf6, 0x9b, 0xb1, 0xcb, 0xa3, 0x89, 0xcf, 0x7d, 0x3e, 0x41, 0xc2, 0x66, 0xbf, 0x45,
-	0x84, 0x00, 0xa7, 0x42, 0x68, 0xfc, 0x80, 0x96, 0xc5, 0x92, 0x5d, 0xe8, 0x3a, 0x54, 0x83, 0x96,
-	0x1d, 0x38, 0xa9, 0xb7, 0x98, 0x6b, 0x64, 0x40, 0x86, 0x4d, 0xeb, 0x02, 0xe9, 0x6b, 0xe8, 0x08,
-	0xd2, 0x62, 0xae, 0x35, 0x70, 0x77, 0xfd, 0x40, 0x9f, 0xc3, 0xa3, 0x6f, 0x09, 0x77, 0x03, 0x4d,
-	0xc6, 0x4d, 0x01, 0x8c, 0xbf, 0xd0, 0x9d, 0xf1, 0x78, 0x1b, 0xfa, 0xb3, 0xc0, 0x89, 0x7d, 0x46,
-	0xdf, 0x97, 0x46, 0x78, 0x5d, 0x31, 0x9f, 0x8d, 0xcb, 0x0e, 0x62, 0x31, 0x6d, 0x1e, 0xff, 0xbf,
-	0x91, 0xac, 0x32, 0xd0, 0x27, 0x80, 0x42, 0xbc, 0xfa, 0x93, 0x30, 0xf4, 0xed, 0x99, 0xfd, 0xab,
-	0xaa, 0x7a, 0x3e, 0x67, 0x58, 0x15, 0xb6, 0xf1, 0x8f, 0xc0, 0x53, 0xdb, 0x0d, 0x98, 0xb7, 0xdf,
-	0xb1, 0x19, 0x8f, 0x22, 0x27, 0xf6, 0x28, 0x85, 0xe6, 0x7a, 0x2d, 0xda, 0x75, 0x2c, 0x9c, 0xe9,
-	0x97, 0x7a, 0x4c, 0x74, 0x51, 0xcc, 0x97, 0x0f, 0xbb, 0x88, 0x80, 0xf5, 0x62, 0x1f, 0x41, 0xb1,
-	0x59, 0x7a, 0x08, 0xdd, 0x22, 0xa6, 0x8c, 0x31, 0x5f, 0x5c, 0x0f, 0x54, 0x96, 0x56, 0x95, 0x69,
-	0xfc, 0xcc, 0xad, 0x31, 0xd9, 0xd4, 0xc9, 0xdc, 0x20, 0x8f, 0xb7, 0x62, 0x69, 0x24, 0x1e, 0x1f,
-	0x67, 0xfa, 0x19, 0xda, 0x82, 0xf3, 0x5b, 0x6b, 0x0c, 0xe4, 0xa1, 0x62, 0xbe, 0xaa, 0x5c, 0xae,
-	0xf7, 0x13, 0xe9, 0x4a, 0xc1, 0x68, 0x09, 0xea, 0xfd, 0x37, 0xa2, 0x0a, 0xb4, 0xbe, 0x7a, 0xde,
-	0x77, 0xee, 0x31, 0x55, 0xa2, 0x3d, 0x00, 0x8b, 0x45, 0xfc, 0xc0, 0x10, 0x13, 0xfa, 0x04, 0x3a,
-	0x76, 0xe6, 0xa4, 0x19, 0xc2, 0x06, 0xed, 0x42, 0xdb, 0xce, 0x78, 0x82, 0x48, 0x1e, 0xbd, 0xad,
-	0xf5, 0xcc, 0xb5, 0x4b, 0xee, 0x8b, 0x2f, 0xaa, 0x94, 0x6b, 0xe7, 0xf1, 0x05, 0x92, 0xa9, 0x7a,
-	0xba, 0xd5, 0xa5, 0xe3, 0x59, 0x27, 0xa7, 0xb3, 0x4e, 0x6e, 0xce, 0x3a, 0xd9, 0x3c, 0xc6, 0x1f,
-	0xee, 0xc3, 0x5d, 0x00, 0x00, 0x00, 0xff, 0xff, 0x5d, 0x25, 0xdc, 0x25, 0xbb, 0x02, 0x00, 0x00,
+func (m *DNStoreInfo) Reset()         { *m = DNStoreInfo{} }
+func (m *DNStoreInfo) String() string { return proto.CompactTextString(m) }
+func (*DNStoreInfo) ProtoMessage()    {}
+func (*DNStoreInfo) Descriptor() ([]byte, []int) {
+	return fileDescriptor_5e1506f3aa5330eb, []int{4}
 }
-
-func (m *Replica) Marshal() (dAtA []byte, err error) {
-	size := m.Size()
-	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
-	if err != nil {
-		return nil, err
+func (m *DNStoreInfo) XXX_Unmarshal(b []byte) error {
+	return m.Unmarshal(b)
+}
+func (m *DNStoreInfo) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	if deterministic {
+		return xxx_messageInfo_DNStoreInfo.Marshal(b, m, deterministic)
+	} else {
+		b = b[:cap(b)]
+		n, err := m.MarshalToSizedBuffer(b)
+		if err != nil {
+			return nil, err
+		}
+		return b[:n], nil
 	}
-	return dAtA[:n], nil
 }
-
-func (m *Replica) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
+func (m *DNStoreInfo) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_DNStoreInfo.Merge(m, src)
+}
+func (m *DNStoreInfo) XXX_Size() int {
+	return m.Size()
+}
+func (m *DNStoreInfo) XXX_DiscardUnknown() {
+	xxx_messageInfo_DNStoreInfo.DiscardUnknown(m)
 }
 
-func (m *Replica) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
-	_ = i
-	var l int
-	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
-	}
-	if m.Epoch != 0 {
-		i = encodeVarintHakeeper(dAtA, i, uint64(m.Epoch))
-		i--
-		dAtA[i] = 0x18
-	}
-	if m.ReplicaID != 0 {
-		i = encodeVarintHakeeper(dAtA, i, uint64(m.ReplicaID))
-		i--
-		dAtA[i] = 0x10
-	}
-	if m.ShardID != 0 {
-		i = encodeVarintHakeeper(dAtA, i, uint64(m.ShardID))
-		i--
-		dAtA[i] = 0x8
+var xxx_messageInfo_DNStoreInfo proto.InternalMessageInfo
+
+func (m *DNStoreInfo) GetTick() uint64 {
+	if m != nil {
+		return m.Tick
 	}
-	return len(dAtA) - i, nil
+	return 0
 }
 
-func (m *ConfigChange) Marshal() (dAtA []byte, err error) {
-	size := m.Size()
-	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
-	if err != nil {
-		return nil, err
+func (m *DNStoreInfo) GetShards() []logservice.DNShardInfo {
+	if m != nil {
+		return m.Shards
 	}
-	return dAtA[:n], nil
+	return nil
 }
 
-func (m *ConfigChange) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
+// DNState contains all DN details known to the HAKeeper.
+type DNState struct {
+	// Stores is keyed by DN store UUID, it contains details found on each DN
+	// store. Each DNStoreInfo reflects what was last reported by each DN store.
+	Stores               map[string]DNStoreInfo `protobuf:"bytes,1,rep,name=Stores,proto3" json:"Stores" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"`
+	XXX_NoUnkeyedLiteral struct{}               `json:"-"`
+	XXX_unrecognized     []byte                 `json:"-"`
+	XXX_sizecache        int32                  `json:"-"`
 }
 
-func (m *ConfigChange) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
-	_ = i
-	var l int
-	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
-	}
-	if m.ChangeType != 0 {
-		i = encodeVarintHakeeper(dAtA, i, uint64(m.ChangeType))
-		i--
-		dAtA[i] = 0x10
-	}
-	{
-		size, err := m.Replica.MarshalToSizedBuffer(dAtA[:i])
+func (m *DNState) Reset()         { *m = DNState{} }
+func (m *DNState) String() string { return proto.CompactTextString(m) }
+func (*DNState) ProtoMessage()    {}
+func (*DNState) Descriptor() ([]byte, []int) {
+	return fileDescriptor_5e1506f3aa5330eb, []int{5}
+}
+func (m *DNState) XXX_Unmarshal(b []byte) error {
+	return m.Unmarshal(b)
+}
+func (m *DNState) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	if deterministic {
+		return xxx_messageInfo_DNState.Marshal(b, m, deterministic)
+	} else {
+		b = b[:cap(b)]
+		n, err := m.MarshalToSizedBuffer(b)
 		if err != nil {
-			return 0, err
+			return nil, err
 		}
-		i -= size
-		i = encodeVarintHakeeper(dAtA, i, uint64(size))
+		return b[:n], nil
 	}
-	i--
-	dAtA[i] = 0xa
-	return len(dAtA) - i, nil
+}
+func (m *DNState) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_DNState.Merge(m, src)
+}
+func (m *DNState) XXX_Size() int {
+	return m.Size()
+}
+func (m *DNState) XXX_DiscardUnknown() {
+	xxx_messageInfo_DNState.DiscardUnknown(m)
 }
 
-func (m *ScheduleCommand) Marshal() (dAtA []byte, err error) {
-	size := m.Size()
-	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
-	if err != nil {
-		return nil, err
+var xxx_messageInfo_DNState proto.InternalMessageInfo
+
+func (m *DNState) GetStores() map[string]DNStoreInfo {
+	if m != nil {
+		return m.Stores
 	}
-	return dAtA[:n], nil
+	return nil
 }
 
-func (m *ScheduleCommand) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
+// ClusterInfo provides a global view of all shards in the cluster. It
+// describes the logical sharding of the system, rather than physical
+// distribution of all replicas that belong to those shards.
+type ClusterInfo struct {
+	DNShards             []metadata.DNShardRecord  `protobuf:"bytes,1,rep,name=DNShards,proto3" json:"DNShards"`
+	LogShards            []metadata.LogShardRecord `protobuf:"bytes,2,rep,name=LogShards,proto3" json:"LogShards"`
+	XXX_NoUnkeyedLiteral struct{}                  `json:"-"`
+	XXX_unrecognized     []byte                    `json:"-"`
+	XXX_sizecache        int32                     `json:"-"`
 }
 
-func (m *ScheduleCommand) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
-	_ = i
-	var l int
-	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
-	}
-	if m.ServiceType != 0 {
-		i = encodeVarintHakeeper(dAtA, i, uint64(m.ServiceType))
-		i--
-		dAtA[i] = 0x18
-	}
-	{
-		size, err := m.ConfigChange.MarshalToSizedBuffer(dAtA[:i])
+func (m *ClusterInfo) Reset()         { *m = ClusterInfo{} }
+func (m *ClusterInfo) String() string { return proto.CompactTextString(m) }
+func (*ClusterInfo) ProtoMessage()    {}
+func (*ClusterInfo) Descriptor() ([]byte, []int) {
+	return fileDescriptor_5e1506f3aa5330eb, []int{6}
+}
+func (m *ClusterInfo) XXX_Unmarshal(b []byte) error {
+	return m.Unmarshal(b)
+}
+func (m *ClusterInfo) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	if deterministic {
+		return xxx_messageInfo_ClusterInfo.Marshal(b, m, deterministic)
+	} else {
+		b = b[:cap(b)]
+		n, err := m.MarshalToSizedBuffer(b)
 		if err != nil {
-			return 0, err
+			return nil, err
 		}
-		i -= size
-		i = encodeVarintHakeeper(dAtA, i, uint64(size))
+		return b[:n], nil
 	}
-	i--
-	dAtA[i] = 0x12
-	if len(m.UUID) > 0 {
-		i -= len(m.UUID)
-		copy(dAtA[i:], m.UUID)
-		i = encodeVarintHakeeper(dAtA, i, uint64(len(m.UUID)))
-		i--
-		dAtA[i] = 0xa
+}
+func (m *ClusterInfo) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_ClusterInfo.Merge(m, src)
+}
+func (m *ClusterInfo) XXX_Size() int {
+	return m.Size()
+}
+func (m *ClusterInfo) XXX_DiscardUnknown() {
+	xxx_messageInfo_ClusterInfo.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_ClusterInfo proto.InternalMessageInfo
+
+func (m *ClusterInfo) GetDNShards() []metadata.DNShardRecord {
+	if m != nil {
+		return m.DNShards
 	}
-	return len(dAtA) - i, nil
+	return nil
 }
 
-func (m *CommandBatch) Marshal() (dAtA []byte, err error) {
-	size := m.Size()
-	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
-	if err != nil {
-		return nil, err
+func (m *ClusterInfo) GetLogShards() []metadata.LogShardRecord {
+	if m != nil {
+		return m.LogShards
 	}
-	return dAtA[:n], nil
+	return nil
 }
 
-func (m *CommandBatch) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
+// LogStoreInfo contains information of all replicas found on a Log store.
+type LogStoreInfo struct {
+	Tick                 uint64                      `protobuf:"varint,1,opt,name=Tick,proto3" json:"Tick,omitempty"`
+	RaftAddress          string                      `protobuf:"bytes,2,opt,name=RaftAddress,proto3" json:"RaftAddress,omitempty"`
+	ServiceAddress       string                      `protobuf:"bytes,3,opt,name=ServiceAddress,proto3" json:"ServiceAddress,omitempty"`
+	GossipAddress        string                      `protobuf:"bytes,4,opt,name=GossipAddress,proto3" json:"GossipAddress,omitempty"`
+	Replicas             []logservice.LogReplicaInfo `protobuf:"bytes,5,rep,name=Replicas,proto3" json:"Replicas"`
+	XXX_NoUnkeyedLiteral struct{}                    `json:"-"`
+	XXX_unrecognized     []byte                      `json:"-"`
+	XXX_sizecache        int32                       `json:"-"`
 }
 
-func (m *CommandBatch) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
-	_ = i
-	var l int
-	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
-	}
-	if len(m.Commands) > 0 {
-		for iNdEx := len(m.Commands) - 1; iNdEx >= 0; iNdEx-- {
-			{
-				size, err := m.Commands[iNdEx].MarshalToSizedBuffer(dAtA[:i])
-				if err != nil {
-					return 0, err
-				}
-				i -= size
-				i = encodeVarintHakeeper(dAtA, i, uint64(size))
-			}
-			i--
-			dAtA[i] = 0x12
+func (m *LogStoreInfo) Reset()         { *m = LogStoreInfo{} }
+func (m *LogStoreInfo) String() string { return proto.CompactTextString(m) }
+func (*LogStoreInfo) ProtoMessage()    {}
+func (*LogStoreInfo) Descriptor() ([]byte, []int) {
+	return fileDescriptor_5e1506f3aa5330eb, []int{7}
+}
+func (m *LogStoreInfo) XXX_Unmarshal(b []byte) error {
+	return m.Unmarshal(b)
+}
+func (m *LogStoreInfo) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	if deterministic {
+		return xxx_messageInfo_LogStoreInfo.Marshal(b, m, deterministic)
+	} else {
+		b = b[:cap(b)]
+		n, err := m.MarshalToSizedBuffer(b)
+		if err != nil {
+			return nil, err
 		}
+		return b[:n], nil
 	}
-	if m.Term != 0 {
-		i = encodeVarintHakeeper(dAtA, i, uint64(m.Term))
-		i--
-		dAtA[i] = 0x8
+}
+func (m *LogStoreInfo) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_LogStoreInfo.Merge(m, src)
+}
+func (m *LogStoreInfo) XXX_Size() int {
+	return m.Size()
+}
+func (m *LogStoreInfo) XXX_DiscardUnknown() {
+	xxx_messageInfo_LogStoreInfo.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_LogStoreInfo proto.InternalMessageInfo
+
+func (m *LogStoreInfo) GetTick() uint64 {
+	if m != nil {
+		return m.Tick
 	}
-	return len(dAtA) - i, nil
+	return 0
 }
 
-func encodeVarintHakeeper(dAtA []byte, offset int, v uint64) int {
-	offset -= sovHakeeper(v)
-	base := offset
-	for v >= 1<<7 {
-		dAtA[offset] = uint8(v&0x7f | 0x80)
-		v >>= 7
-		offset++
+func (m *LogStoreInfo) GetRaftAddress() string {
+	if m != nil {
+		return m.RaftAddress
 	}
-	dAtA[offset] = uint8(v)
-	return base
+	return ""
 }
-func (m *Replica) Size() (n int) {
-	if m == nil {
-		return 0
+
+func (m *LogStoreInfo) GetServiceAddress() string {
+	if m != nil {
+		return m.ServiceAddress
 	}
-	var l int
-	_ = l
-	if m.ShardID != 0 {
-		n += 1 + sovHakeeper(uint64(m.ShardID))
+	return ""
+}
+
+func (m *LogStoreInfo) GetGossipAddress() string {
+	if m != nil {
+		return m.GossipAddress
 	}
-	if m.ReplicaID != 0 {
-		n += 1 + sovHakeeper(uint64(m.ReplicaID))
+	return ""
+}
+
+func (m *LogStoreInfo) GetReplicas() []logservice.LogReplicaInfo {
+	if m != nil {
+		return m.Replicas
 	}
-	if m.Epoch != 0 {
-		n += 1 + sovHakeeper(uint64(m.Epoch))
+	return nil
+}
+
+type LogState struct {
+	// Shards is keyed by ShardID, it contains details aggregated from all Log
+	// stores. Each pb.LogShardInfo here contains data aggregated from
+	// different replicas and thus reflect a more accurate description on each
+	// shard.
+	Shards map[uint64]logservice.LogShardInfo `protobuf:"bytes,1,rep,name=Shards,proto3" json:"Shards" protobuf_key:"varint,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"`
+	// Stores is keyed by log store UUID, it contains details found on each store.
+	// Each LogStoreInfo here reflects what was last reported by each Log store.
+	Stores               map[string]LogStoreInfo `protobuf:"bytes,2,rep,name=Stores,proto3" json:"Stores" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"`
+	XXX_NoUnkeyedLiteral struct{}                `json:"-"`
+	XXX_unrecognized     []byte                  `json:"-"`
+	XXX_sizecache        int32                   `json:"-"`
+}
+
+func (m *LogState) Reset()         { *m = LogState{} }
+func (m *LogState) String() string { return proto.CompactTextString(m) }
+func (*LogState) ProtoMessage()    {}
+func (*LogState) Descriptor() ([]byte, []int) {
+	return fileDescriptor_5e1506f3aa5330eb, []int{8}
+}
+func (m *LogState) XXX_Unmarshal(b []byte) error {
+	return m.Unmarshal(b)
+}
+func (m *LogState) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	if deterministic {
+		return xxx_messageInfo_LogState.Marshal(b, m, deterministic)
+	} else {
+		b = b[:cap(b)]
+		n, err := m.MarshalToSizedBuffer(b)
+		if err != nil {
+			return nil, err
+		}
+		return b[:n], nil
+	}
+}
+func (m *LogState) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_LogState.Merge(m, src)
+}
+func (m *LogState) XXX_Size() int {
+	return m.Size()
+}
+func (m *LogState) XXX_DiscardUnknown() {
+	xxx_messageInfo_LogState.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_LogState proto.InternalMessageInfo
+
+func (m *LogState) GetShards() map[uint64]logservice.LogShardInfo {
+	if m != nil {
+		return m.Shards
+	}
+	return nil
+}
+
+func (m *LogState) GetStores() map[string]LogStoreInfo {
+	if m != nil {
+		return m.Stores
+	}
+	return nil
+}
+
+// HAKeeperState contains all HAKeeper state required for making schedule
+// commands.
+type HAKeeperState struct {
+	Tick                 uint64      `protobuf:"varint,1,opt,name=Tick,proto3" json:"Tick,omitempty"`
+	ClusterInfo          ClusterInfo `protobuf:"bytes,2,opt,name=ClusterInfo,proto3" json:"ClusterInfo"`
+	DNState              DNState     `protobuf:"bytes,3,opt,name=DNState,proto3" json:"DNState"`
+	LogState             LogState    `protobuf:"bytes,4,opt,name=LogState,proto3" json:"LogState"`
+	XXX_NoUnkeyedLiteral struct{}    `json:"-"`
+	XXX_unrecognized     []byte      `json:"-"`
+	XXX_sizecache        int32       `json:"-"`
+}
+
+func (m *HAKeeperState) Reset()         { *m = HAKeeperState{} }
+func (m *HAKeeperState) String() string { return proto.CompactTextString(m) }
+func (*HAKeeperState) ProtoMessage()    {}
+func (*HAKeeperState) Descriptor() ([]byte, []int) {
+	return fileDescriptor_5e1506f3aa5330eb, []int{9}
+}
+func (m *HAKeeperState) XXX_Unmarshal(b []byte) error {
+	return m.Unmarshal(b)
+}
+func (m *HAKeeperState) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	if deterministic {
+		return xxx_messageInfo_HAKeeperState.Marshal(b, m, deterministic)
+	} else {
+		b = b[:cap(b)]
+		n, err := m.MarshalToSizedBuffer(b)
+		if err != nil {
+			return nil, err
+		}
+		return b[:n], nil
+	}
+}
+func (m *HAKeeperState) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_HAKeeperState.Merge(m, src)
+}
+func (m *HAKeeperState) XXX_Size() int {
+	return m.Size()
+}
+func (m *HAKeeperState) XXX_DiscardUnknown() {
+	xxx_messageInfo_HAKeeperState.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_HAKeeperState proto.InternalMessageInfo
+
+func (m *HAKeeperState) GetTick() uint64 {
+	if m != nil {
+		return m.Tick
+	}
+	return 0
+}
+
+func (m *HAKeeperState) GetClusterInfo() ClusterInfo {
+	if m != nil {
+		return m.ClusterInfo
+	}
+	return ClusterInfo{}
+}
+
+func (m *HAKeeperState) GetDNState() DNState {
+	if m != nil {
+		return m.DNState
+	}
+	return DNState{}
+}
+
+func (m *HAKeeperState) GetLogState() LogState {
+	if m != nil {
+		return m.LogState
+	}
+	return LogState{}
+}
+
+// RSMState contains state maintained by HAKeeper's RSM.
+type RSMState struct {
+	Tick                 uint64                  `protobuf:"varint,1,opt,name=Tick,proto3" json:"Tick,omitempty"`
+	NextID               uint64                  `protobuf:"varint,2,opt,name=NextID,proto3" json:"NextID,omitempty"`
+	Term                 uint64                  `protobuf:"varint,3,opt,name=Term,proto3" json:"Term,omitempty"`
+	ScheduleCommands     map[string]CommandBatch `protobuf:"bytes,4,rep,name=ScheduleCommands,proto3" json:"ScheduleCommands" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"`
+	LogShards            map[string]uint64       `protobuf:"bytes,5,rep,name=LogShards,proto3" json:"LogShards,omitempty" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"varint,2,opt,name=value,proto3"`
+	DNState              DNState                 `protobuf:"bytes,6,opt,name=DNState,proto3" json:"DNState"`
+	LogState             LogState                `protobuf:"bytes,7,opt,name=LogState,proto3" json:"LogState"`
+	ClusterInfo          ClusterInfo             `protobuf:"bytes,8,opt,name=ClusterInfo,proto3" json:"ClusterInfo"`
+	XXX_NoUnkeyedLiteral struct{}                `json:"-"`
+	XXX_unrecognized     []byte                  `json:"-"`
+	XXX_sizecache        int32                   `json:"-"`
+}
+
+func (m *RSMState) Reset()         { *m = RSMState{} }
+func (m *RSMState) String() string { return proto.CompactTextString(m) }
+func (*RSMState) ProtoMessage()    {}
+func (*RSMState) Descriptor() ([]byte, []int) {
+	return fileDescriptor_5e1506f3aa5330eb, []int{10}
+}
+func (m *RSMState) XXX_Unmarshal(b []byte) error {
+	return m.Unmarshal(b)
+}
+func (m *RSMState) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	if deterministic {
+		return xxx_messageInfo_RSMState.Marshal(b, m, deterministic)
+	} else {
+		b = b[:cap(b)]
+		n, err := m.MarshalToSizedBuffer(b)
+		if err != nil {
+			return nil, err
+		}
+		return b[:n], nil
+	}
+}
+func (m *RSMState) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_RSMState.Merge(m, src)
+}
+func (m *RSMState) XXX_Size() int {
+	return m.Size()
+}
+func (m *RSMState) XXX_DiscardUnknown() {
+	xxx_messageInfo_RSMState.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_RSMState proto.InternalMessageInfo
+
+func (m *RSMState) GetTick() uint64 {
+	if m != nil {
+		return m.Tick
+	}
+	return 0
+}
+
+func (m *RSMState) GetNextID() uint64 {
+	if m != nil {
+		return m.NextID
+	}
+	return 0
+}
+
+func (m *RSMState) GetTerm() uint64 {
+	if m != nil {
+		return m.Term
+	}
+	return 0
+}
+
+func (m *RSMState) GetScheduleCommands() map[string]CommandBatch {
+	if m != nil {
+		return m.ScheduleCommands
+	}
+	return nil
+}
+
+func (m *RSMState) GetLogShards() map[string]uint64 {
+	if m != nil {
+		return m.LogShards
+	}
+	return nil
+}
+
+func (m *RSMState) GetDNState() DNState {
+	if m != nil {
+		return m.DNState
+	}
+	return DNState{}
+}
+
+func (m *RSMState) GetLogState() LogState {
+	if m != nil {
+		return m.LogState
+	}
+	return LogState{}
+}
+
+func (m *RSMState) GetClusterInfo() ClusterInfo {
+	if m != nil {
+		return m.ClusterInfo
+	}
+	return ClusterInfo{}
+}
+
+func init() {
+	proto.RegisterEnum("hakeeper.ConfigChangeType", ConfigChangeType_name, ConfigChangeType_value)
+	proto.RegisterEnum("hakeeper.ServiceType", ServiceType_name, ServiceType_value)
+	proto.RegisterType((*Replica)(nil), "hakeeper.Replica")
+	proto.RegisterType((*ConfigChange)(nil), "hakeeper.ConfigChange")
+	proto.RegisterType((*ScheduleCommand)(nil), "hakeeper.ScheduleCommand")
+	proto.RegisterType((*CommandBatch)(nil), "hakeeper.CommandBatch")
+	proto.RegisterType((*DNStoreInfo)(nil), "hakeeper.DNStoreInfo")
+	proto.RegisterType((*DNState)(nil), "hakeeper.DNState")
+	proto.RegisterMapType((map[string]DNStoreInfo)(nil), "hakeeper.DNState.StoresEntry")
+	proto.RegisterType((*ClusterInfo)(nil), "hakeeper.ClusterInfo")
+	proto.RegisterType((*LogStoreInfo)(nil), "hakeeper.LogStoreInfo")
+	proto.RegisterType((*LogState)(nil), "hakeeper.LogState")
+	proto.RegisterMapType((map[uint64]logservice.LogShardInfo)(nil), "hakeeper.LogState.ShardsEntry")
+	proto.RegisterMapType((map[string]LogStoreInfo)(nil), "hakeeper.LogState.StoresEntry")
+	proto.RegisterType((*HAKeeperState)(nil), "hakeeper.HAKeeperState")
+	proto.RegisterType((*RSMState)(nil), "hakeeper.RSMState")
+	proto.RegisterMapType((map[string]uint64)(nil), "hakeeper.RSMState.LogShardsEntry")
+	proto.RegisterMapType((map[string]CommandBatch)(nil), "hakeeper.RSMState.ScheduleCommandsEntry")
+}
+
+func init() { proto.RegisterFile("hakeeper.proto", fileDescriptor_5e1506f3aa5330eb) }
+
+var fileDescriptor_5e1506f3aa5330eb = []byte{
+	// 889 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x9c, 0x56, 0xdd, 0x6e, 0xe3, 0x44,
+	0x14, 0x8e, 0x93, 0x34, 0x71, 0x8e, 0xdb, 0x10, 0x46, 0xdb, 0xae, 0x89, 0x20, 0x14, 0x0b, 0xa1,
+	0x6a, 0x59, 0xbc, 0xda, 0x00, 0x02, 0x76, 0x83, 0xa0, 0x4d, 0x56, 0x50, 0x51, 0x22, 0xb0, 0xb7,
+	0x02, 0xc1, 0x05, 0x72, 0xed, 0x89, 0x13, 0xe5, 0x67, 0x22, 0xdb, 0xa9, 0xe8, 0x05, 0xb7, 0x3c,
+	0x03, 0x97, 0x5c, 0x73, 0xc1, 0x73, 0x54, 0xe2, 0xa6, 0x4f, 0x80, 0xa0, 0x4f, 0x82, 0x7c, 0x3c,
+	0x63, 0xcf, 0xa4, 0x29, 0xd0, 0xbd, 0xf3, 0x99, 0xf3, 0x9d, 0x6f, 0xce, 0xcf, 0x37, 0x33, 0x86,
+	0xe6, 0xd8, 0x9b, 0x52, 0xba, 0xa4, 0x91, 0xbd, 0x8c, 0x58, 0xc2, 0x88, 0x2e, 0xec, 0xf6, 0x3b,
+	0xe1, 0x24, 0x19, 0xaf, 0xce, 0x6c, 0x9f, 0xcd, 0x1f, 0x85, 0x2c, 0x64, 0x8f, 0x10, 0x70, 0xb6,
+	0x1a, 0xa1, 0x85, 0x06, 0x7e, 0x65, 0x81, 0xed, 0xd6, 0x8c, 0x85, 0x31, 0x8d, 0xce, 0x27, 0x3e,
+	0xe5, 0x2b, 0xcd, 0x39, 0x4d, 0xbc, 0xc0, 0x4b, 0xbc, 0xcc, 0xb6, 0xbe, 0x81, 0xba, 0x43, 0x97,
+	0xb3, 0x89, 0xef, 0x11, 0x13, 0xea, 0xee, 0xd8, 0x8b, 0x82, 0xe3, 0x81, 0xa9, 0xed, 0x6b, 0x07,
+	0x55, 0x47, 0x98, 0xe4, 0x55, 0x68, 0x70, 0xd0, 0xf1, 0xc0, 0x2c, 0xa3, 0xaf, 0x58, 0x20, 0xf7,
+	0x60, 0xeb, 0xd9, 0x92, 0xf9, 0x63, 0xb3, 0x82, 0x9e, 0xcc, 0xb0, 0x7e, 0x82, 0xed, 0x3e, 0x5b,
+	0x8c, 0x26, 0x61, 0x7f, 0xec, 0x2d, 0x42, 0x4a, 0x1e, 0xe7, 0x1b, 0x21, 0xbb, 0xd1, 0x7d, 0xd9,
+	0xce, 0xab, 0xe4, 0x8e, 0xa3, 0xea, 0xe5, 0x9f, 0xaf, 0x97, 0x9c, 0x3c, 0xa1, 0x27, 0x00, 0x59,
+	0xf0, 0xf3, 0x8b, 0x25, 0xc5, 0x7d, 0x9b, 0xdd, 0x76, 0x11, 0x25, 0xd3, 0xa7, 0x08, 0x47, 0x42,
+	0x5b, 0xbf, 0x6a, 0xf0, 0x92, 0xeb, 0x8f, 0x69, 0xb0, 0x9a, 0xd1, 0x3e, 0x9b, 0xcf, 0xbd, 0x45,
+	0x40, 0x08, 0x54, 0x4f, 0x4f, 0x79, 0x75, 0x0d, 0x07, 0xbf, 0xc9, 0xa7, 0x6a, 0x9a, 0xb8, 0x8b,
+	0xd1, 0xdd, 0xdb, 0xbc, 0x0b, 0x4f, 0x50, 0x2d, 0xec, 0x03, 0x30, 0xdc, 0xac, 0xc5, 0x98, 0x66,
+	0x05, 0xd3, 0xdc, 0x2d, 0x08, 0x24, 0xa7, 0x23, 0x23, 0xad, 0x1f, 0xd2, 0xad, 0x31, 0xb3, 0x23,
+	0x2f, 0xf1, 0xc7, 0x69, 0x7a, 0xcf, 0x69, 0x34, 0xe7, 0xcd, 0xc7, 0x6f, 0xf2, 0x14, 0x74, 0x8e,
+	0x89, 0xcd, 0xf2, 0x7e, 0xe5, 0xc0, 0xe8, 0xbe, 0x22, 0x31, 0xab, 0xf5, 0xf1, 0xec, 0xf2, 0x00,
+	0xeb, 0x5b, 0x30, 0x06, 0x43, 0x37, 0x61, 0x11, 0x3d, 0x5e, 0x8c, 0x18, 0xf2, 0x4f, 0xfc, 0x69,
+	0xce, 0x3f, 0xf1, 0xa7, 0xe4, 0x7d, 0xa8, 0xe1, 0x90, 0x05, 0xfb, 0x7d, 0x5b, 0x52, 0xcc, 0x60,
+	0x98, 0x09, 0x60, 0x31, 0x62, 0x9c, 0x9b, 0x83, 0xad, 0x5f, 0x34, 0xa8, 0xa7, 0xd4, 0x5e, 0x42,
+	0xc9, 0x53, 0xa8, 0xe1, 0x1e, 0xb1, 0xa9, 0x21, 0xc5, 0x6b, 0x45, 0x82, 0x1c, 0x62, 0x67, 0xfe,
+	0x67, 0x8b, 0x24, 0xba, 0xc8, 0x89, 0x70, 0xa9, 0xfd, 0x15, 0x18, 0x92, 0x93, 0xb4, 0xa0, 0x32,
+	0xa5, 0x17, 0x7c, 0x40, 0xe9, 0x27, 0x79, 0x1b, 0xb6, 0xce, 0xbd, 0xd9, 0x4a, 0x0c, 0x66, 0x57,
+	0x25, 0xe7, 0xa5, 0x39, 0x19, 0xe6, 0x49, 0xf9, 0x43, 0xcd, 0xfa, 0x59, 0x03, 0xa3, 0x3f, 0x5b,
+	0xc5, 0x09, 0x8d, 0xb0, 0xea, 0x8f, 0x40, 0xe7, 0x75, 0x88, 0x04, 0xef, 0xdb, 0xf9, 0x19, 0xe0,
+	0x1e, 0x87, 0xfa, 0x2c, 0xca, 0xfb, 0x27, 0xe0, 0xa4, 0x07, 0x8d, 0x13, 0x16, 0x2a, 0xfd, 0x31,
+	0x8b, 0x58, 0xe1, 0x52, 0x82, 0x8b, 0x00, 0xeb, 0x4a, 0x83, 0xed, 0xd4, 0xfa, 0xd7, 0xfe, 0xef,
+	0x83, 0xe1, 0x78, 0xa3, 0xe4, 0x30, 0x08, 0x22, 0x1a, 0xc7, 0x58, 0x64, 0xc3, 0x91, 0x97, 0xc8,
+	0x5b, 0xd0, 0xe4, 0xa2, 0x11, 0xa0, 0x0a, 0x82, 0xd6, 0x56, 0xc9, 0x9b, 0xb0, 0xf3, 0x19, 0x8b,
+	0xe3, 0xc9, 0x52, 0xc0, 0xaa, 0x08, 0x53, 0x17, 0x49, 0x0f, 0x74, 0x7e, 0xba, 0x62, 0x73, 0x0b,
+	0x2b, 0x6a, 0xcb, 0x13, 0x3f, 0x61, 0xa1, 0x38, 0xd7, 0xc5, 0xd0, 0xf3, 0x08, 0xeb, 0xf7, 0x32,
+	0xe8, 0x58, 0x52, 0x3a, 0xf7, 0x5e, 0x2e, 0x9d, 0xac, 0xad, 0x9d, 0x62, 0x34, 0x02, 0x63, 0x67,
+	0x00, 0x75, 0xf0, 0xa2, 0xb7, 0x42, 0x35, 0xe5, 0xdb, 0xa3, 0x6f, 0x95, 0x8d, 0x0b, 0x86, 0x44,
+	0x2d, 0xcb, 0xa6, 0x9a, 0xc9, 0xc6, 0x56, 0x65, 0x63, 0xae, 0x15, 0x99, 0xeb, 0x5a, 0x52, 0x4e,
+	0xfb, 0xeb, 0xff, 0xd2, 0xe2, 0x43, 0x95, 0x74, 0x6f, 0x2d, 0xe5, 0x0d, 0x62, 0xfc, 0x43, 0x83,
+	0x9d, 0xcf, 0x0f, 0xbf, 0x40, 0x50, 0xd6, 0xb5, 0x4d, 0x22, 0xf8, 0x58, 0x51, 0xec, 0x4d, 0xa5,
+	0x4b, 0x4e, 0xde, 0x07, 0x45, 0xe1, 0x8f, 0xf3, 0xb3, 0x88, 0xd2, 0x50, 0x6e, 0x56, 0xee, 0x10,
+	0x37, 0xab, 0x38, 0xb3, 0xef, 0x15, 0x73, 0x44, 0x9d, 0x18, 0x5d, 0x72, 0xb3, 0xff, 0x62, 0xfc,
+	0xc2, 0xb6, 0x7e, 0xab, 0x82, 0xee, 0xb8, 0x5f, 0xde, 0x5e, 0xc8, 0x1e, 0xd4, 0x86, 0xf4, 0xc7,
+	0x24, 0x7f, 0x24, 0xb8, 0x95, 0xdf, 0x6c, 0x15, 0xe9, 0x66, 0xfb, 0x0e, 0x5a, 0x6b, 0xf7, 0x57,
+	0x2a, 0xd9, 0x54, 0x0a, 0x07, 0xd2, 0xc3, 0xc0, 0x77, 0x5b, 0xbf, 0xea, 0x14, 0x51, 0xdc, 0xe0,
+	0x21, 0x9f, 0xc8, 0x07, 0x37, 0x93, 0xf9, 0x1b, 0x1b, 0x48, 0x73, 0x0c, 0xb2, 0x49, 0x67, 0x57,
+	0x6e, 0x69, 0xed, 0x05, 0x5a, 0x5a, 0xff, 0xbf, 0x2d, 0x5d, 0x1f, 0xbd, 0x7e, 0xb7, 0xd1, 0xb7,
+	0xbf, 0x87, 0xdd, 0x8d, 0x9d, 0xb9, 0x93, 0x78, 0xe5, 0x47, 0x48, 0x3e, 0x0f, 0x3d, 0x68, 0xaa,
+	0x1d, 0xda, 0xc0, 0x7a, 0x4f, 0x66, 0xad, 0x4a, 0xd1, 0x0f, 0x4e, 0xa0, 0xb5, 0xfe, 0x40, 0x13,
+	0x03, 0xea, 0x87, 0x41, 0x30, 0x64, 0x01, 0x6d, 0x95, 0x48, 0x13, 0xc0, 0xa1, 0x73, 0x76, 0x4e,
+	0xd1, 0xd6, 0xc8, 0x0e, 0x34, 0xdc, 0xc4, 0x8b, 0x12, 0x34, 0xcb, 0x64, 0x1b, 0x74, 0x37, 0x61,
+	0x4b, 0xb4, 0x2a, 0x0f, 0x1e, 0x2a, 0x8f, 0x6c, 0x1a, 0x9b, 0xa6, 0x96, 0xad, 0xb4, 0x4a, 0x69,
+	0xec, 0x60, 0x21, 0x4c, 0xed, 0xa8, 0x75, 0xf5, 0x77, 0xa7, 0x74, 0x79, 0xdd, 0xd1, 0xae, 0xae,
+	0x3b, 0xda, 0x5f, 0xd7, 0x1d, 0xed, 0xac, 0x86, 0x7f, 0x3b, 0xef, 0xfe, 0x13, 0x00, 0x00, 0xff,
+	0xff, 0x9f, 0x72, 0x4d, 0x54, 0x5a, 0x09, 0x00, 0x00,
+}
+
+func (m *Replica) Marshal() (dAtA []byte, err error) {
+	size := m.Size()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	if err != nil {
+		return nil, err
 	}
+	return dAtA[:n], nil
+}
+
+func (m *Replica) MarshalTo(dAtA []byte) (int, error) {
+	size := m.Size()
+	return m.MarshalToSizedBuffer(dAtA[:size])
+}
+
+func (m *Replica) MarshalToSizedBuffer(dAtA []byte) (int, error) {
+	i := len(dAtA)
+	_ = i
+	var l int
+	_ = l
 	if m.XXX_unrecognized != nil {
-		n += len(m.XXX_unrecognized)
+		i -= len(m.XXX_unrecognized)
+		copy(dAtA[i:], m.XXX_unrecognized)
 	}
-	return n
+	if m.Epoch != 0 {
+		i = encodeVarintHakeeper(dAtA, i, uint64(m.Epoch))
+		i--
+		dAtA[i] = 0x18
+	}
+	if m.ReplicaID != 0 {
+		i = encodeVarintHakeeper(dAtA, i, uint64(m.ReplicaID))
+		i--
+		dAtA[i] = 0x10
+	}
+	if m.ShardID != 0 {
+		i = encodeVarintHakeeper(dAtA, i, uint64(m.ShardID))
+		i--
+		dAtA[i] = 0x8
+	}
+	return len(dAtA) - i, nil
 }
 
-func (m *ConfigChange) Size() (n int) {
-	if m == nil {
-		return 0
+func (m *ConfigChange) Marshal() (dAtA []byte, err error) {
+	size := m.Size()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	if err != nil {
+		return nil, err
 	}
+	return dAtA[:n], nil
+}
+
+func (m *ConfigChange) MarshalTo(dAtA []byte) (int, error) {
+	size := m.Size()
+	return m.MarshalToSizedBuffer(dAtA[:size])
+}
+
+func (m *ConfigChange) MarshalToSizedBuffer(dAtA []byte) (int, error) {
+	i := len(dAtA)
+	_ = i
 	var l int
 	_ = l
-	l = m.Replica.Size()
-	n += 1 + l + sovHakeeper(uint64(l))
+	if m.XXX_unrecognized != nil {
+		i -= len(m.XXX_unrecognized)
+		copy(dAtA[i:], m.XXX_unrecognized)
+	}
 	if m.ChangeType != 0 {
-		n += 1 + sovHakeeper(uint64(m.ChangeType))
+		i = encodeVarintHakeeper(dAtA, i, uint64(m.ChangeType))
+		i--
+		dAtA[i] = 0x10
+	}
+	{
+		size, err := m.Replica.MarshalToSizedBuffer(dAtA[:i])
+		if err != nil {
+			return 0, err
+		}
+		i -= size
+		i = encodeVarintHakeeper(dAtA, i, uint64(size))
+	}
+	i--
+	dAtA[i] = 0xa
+	return len(dAtA) - i, nil
+}
+
+func (m *ScheduleCommand) Marshal() (dAtA []byte, err error) {
+	size := m.Size()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	if err != nil {
+		return nil, err
 	}
+	return dAtA[:n], nil
+}
+
+func (m *ScheduleCommand) MarshalTo(dAtA []byte) (int, error) {
+	size := m.Size()
+	return m.MarshalToSizedBuffer(dAtA[:size])
+}
+
+func (m *ScheduleCommand) MarshalToSizedBuffer(dAtA []byte) (int, error) {
+	i := len(dAtA)
+	_ = i
+	var l int
+	_ = l
 	if m.XXX_unrecognized != nil {
-		n += len(m.XXX_unrecognized)
+		i -= len(m.XXX_unrecognized)
+		copy(dAtA[i:], m.XXX_unrecognized)
+	}
+	if m.ServiceType != 0 {
+		i = encodeVarintHakeeper(dAtA, i, uint64(m.ServiceType))
+		i--
+		dAtA[i] = 0x18
+	}
+	{
+		size, err := m.ConfigChange.MarshalToSizedBuffer(dAtA[:i])
+		if err != nil {
+			return 0, err
+		}
+		i -= size
+		i = encodeVarintHakeeper(dAtA, i, uint64(size))
+	}
+	i--
+	dAtA[i] = 0x12
+	if len(m.UUID) > 0 {
+		i -= len(m.UUID)
+		copy(dAtA[i:], m.UUID)
+		i = encodeVarintHakeeper(dAtA, i, uint64(len(m.UUID)))
+		i--
+		dAtA[i] = 0xa
+	}
+	return len(dAtA) - i, nil
+}
+
+func (m *CommandBatch) Marshal() (dAtA []byte, err error) {
+	size := m.Size()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	if err != nil {
+		return nil, err
+	}
+	return dAtA[:n], nil
+}
+
+func (m *CommandBatch) MarshalTo(dAtA []byte) (int, error) {
+	size := m.Size()
+	return m.MarshalToSizedBuffer(dAtA[:size])
+}
+
+func (m *CommandBatch) MarshalToSizedBuffer(dAtA []byte) (int, error) {
+	i := len(dAtA)
+	_ = i
+	var l int
+	_ = l
+	if m.XXX_unrecognized != nil {
+		i -= len(m.XXX_unrecognized)
+		copy(dAtA[i:], m.XXX_unrecognized)
+	}
+	if len(m.Commands) > 0 {
+		for iNdEx := len(m.Commands) - 1; iNdEx >= 0; iNdEx-- {
+			{
+				size, err := m.Commands[iNdEx].MarshalToSizedBuffer(dAtA[:i])
+				if err != nil {
+					return 0, err
+				}
+				i -= size
+				i = encodeVarintHakeeper(dAtA, i, uint64(size))
+			}
+			i--
+			dAtA[i] = 0x12
+		}
+	}
+	if m.Term != 0 {
+		i = encodeVarintHakeeper(dAtA, i, uint64(m.Term))
+		i--
+		dAtA[i] = 0x8
+	}
+	return len(dAtA) - i, nil
+}
+
+func (m *DNStoreInfo) Marshal() (dAtA []byte, err error) {
+	size := m.Size()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	if err != nil {
+		return nil, err
+	}
+	return dAtA[:n], nil
+}
+
+func (m *DNStoreInfo) MarshalTo(dAtA []byte) (int, error) {
+	size := m.Size()
+	return m.MarshalToSizedBuffer(dAtA[:size])
+}
+
+func (m *DNStoreInfo) MarshalToSizedBuffer(dAtA []byte) (int, error) {
+	i := len(dAtA)
+	_ = i
+	var l int
+	_ = l
+	if m.XXX_unrecognized != nil {
+		i -= len(m.XXX_unrecognized)
+		copy(dAtA[i:], m.XXX_unrecognized)
+	}
+	if len(m.Shards) > 0 {
+		for iNdEx := len(m.Shards) - 1; iNdEx >= 0; iNdEx-- {
+			{
+				size, err := m.Shards[iNdEx].MarshalToSizedBuffer(dAtA[:i])
+				if err != nil {
+					return 0, err
+				}
+				i -= size
+				i = encodeVarintHakeeper(dAtA, i, uint64(size))
+			}
+			i--
+			dAtA[i] = 0x12
+		}
+	}
+	if m.Tick != 0 {
+		i = encodeVarintHakeeper(dAtA, i, uint64(m.Tick))
+		i--
+		dAtA[i] = 0x8
+	}
+	return len(dAtA) - i, nil
+}
+
+func (m *DNState) Marshal() (dAtA []byte, err error) {
+	size := m.Size()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	if err != nil {
+		return nil, err
+	}
+	return dAtA[:n], nil
+}
+
+func (m *DNState) MarshalTo(dAtA []byte) (int, error) {
+	size := m.Size()
+	return m.MarshalToSizedBuffer(dAtA[:size])
+}
+
+func (m *DNState) MarshalToSizedBuffer(dAtA []byte) (int, error) {
+	i := len(dAtA)
+	_ = i
+	var l int
+	_ = l
+	if m.XXX_unrecognized != nil {
+		i -= len(m.XXX_unrecognized)
+		copy(dAtA[i:], m.XXX_unrecognized)
+	}
+	if len(m.Stores) > 0 {
+		for k := range m.Stores {
+			v := m.Stores[k]
+			baseI := i
+			{
+				size, err := (&v).MarshalToSizedBuffer(dAtA[:i])
+				if err != nil {
+					return 0, err
+				}
+				i -= size
+				i = encodeVarintHakeeper(dAtA, i, uint64(size))
+			}
+			i--
+			dAtA[i] = 0x12
+			i -= len(k)
+			copy(dAtA[i:], k)
+			i = encodeVarintHakeeper(dAtA, i, uint64(len(k)))
+			i--
+			dAtA[i] = 0xa
+			i = encodeVarintHakeeper(dAtA, i, uint64(baseI-i))
+			i--
+			dAtA[i] = 0xa
+		}
+	}
+	return len(dAtA) - i, nil
+}
+
+func (m *ClusterInfo) Marshal() (dAtA []byte, err error) {
+	size := m.Size()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	if err != nil {
+		return nil, err
+	}
+	return dAtA[:n], nil
+}
+
+func (m *ClusterInfo) MarshalTo(dAtA []byte) (int, error) {
+	size := m.Size()
+	return m.MarshalToSizedBuffer(dAtA[:size])
+}
+
+func (m *ClusterInfo) MarshalToSizedBuffer(dAtA []byte) (int, error) {
+	i := len(dAtA)
+	_ = i
+	var l int
+	_ = l
+	if m.XXX_unrecognized != nil {
+		i -= len(m.XXX_unrecognized)
+		copy(dAtA[i:], m.XXX_unrecognized)
+	}
+	if len(m.LogShards) > 0 {
+		for iNdEx := len(m.LogShards) - 1; iNdEx >= 0; iNdEx-- {
+			{
+				size, err := m.LogShards[iNdEx].MarshalToSizedBuffer(dAtA[:i])
+				if err != nil {
+					return 0, err
+				}
+				i -= size
+				i = encodeVarintHakeeper(dAtA, i, uint64(size))
+			}
+			i--
+			dAtA[i] = 0x12
+		}
+	}
+	if len(m.DNShards) > 0 {
+		for iNdEx := len(m.DNShards) - 1; iNdEx >= 0; iNdEx-- {
+			{
+				size, err := m.DNShards[iNdEx].MarshalToSizedBuffer(dAtA[:i])
+				if err != nil {
+					return 0, err
+				}
+				i -= size
+				i = encodeVarintHakeeper(dAtA, i, uint64(size))
+			}
+			i--
+			dAtA[i] = 0xa
+		}
+	}
+	return len(dAtA) - i, nil
+}
+
+func (m *LogStoreInfo) Marshal() (dAtA []byte, err error) {
+	size := m.Size()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	if err != nil {
+		return nil, err
+	}
+	return dAtA[:n], nil
+}
+
+func (m *LogStoreInfo) MarshalTo(dAtA []byte) (int, error) {
+	size := m.Size()
+	return m.MarshalToSizedBuffer(dAtA[:size])
+}
+
+func (m *LogStoreInfo) MarshalToSizedBuffer(dAtA []byte) (int, error) {
+	i := len(dAtA)
+	_ = i
+	var l int
+	_ = l
+	if m.XXX_unrecognized != nil {
+		i -= len(m.XXX_unrecognized)
+		copy(dAtA[i:], m.XXX_unrecognized)
+	}
+	if len(m.Replicas) > 0 {
+		for iNdEx := len(m.Replicas) - 1; iNdEx >= 0; iNdEx-- {
+			{
+				size, err := m.Replicas[iNdEx].MarshalToSizedBuffer(dAtA[:i])
+				if err != nil {
+					return 0, err
+				}
+				i -= size
+				i = encodeVarintHakeeper(dAtA, i, uint64(size))
+			}
+			i--
+			dAtA[i] = 0x2a
+		}
+	}
+	if len(m.GossipAddress) > 0 {
+		i -= len(m.GossipAddress)
+		copy(dAtA[i:], m.GossipAddress)
+		i = encodeVarintHakeeper(dAtA, i, uint64(len(m.GossipAddress)))
+		i--
+		dAtA[i] = 0x22
+	}
+	if len(m.ServiceAddress) > 0 {
+		i -= len(m.ServiceAddress)
+		copy(dAtA[i:], m.ServiceAddress)
+		i = encodeVarintHakeeper(dAtA, i, uint64(len(m.ServiceAddress)))
+		i--
+		dAtA[i] = 0x1a
+	}
+	if len(m.RaftAddress) > 0 {
+		i -= len(m.RaftAddress)
+		copy(dAtA[i:], m.RaftAddress)
+		i = encodeVarintHakeeper(dAtA, i, uint64(len(m.RaftAddress)))
+		i--
+		dAtA[i] = 0x12
+	}
+	if m.Tick != 0 {
+		i = encodeVarintHakeeper(dAtA, i, uint64(m.Tick))
+		i--
+		dAtA[i] = 0x8
+	}
+	return len(dAtA) - i, nil
+}
+
+func (m *LogState) Marshal() (dAtA []byte, err error) {
+	size := m.Size()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	if err != nil {
+		return nil, err
+	}
+	return dAtA[:n], nil
+}
+
+func (m *LogState) MarshalTo(dAtA []byte) (int, error) {
+	size := m.Size()
+	return m.MarshalToSizedBuffer(dAtA[:size])
+}
+
+func (m *LogState) MarshalToSizedBuffer(dAtA []byte) (int, error) {
+	i := len(dAtA)
+	_ = i
+	var l int
+	_ = l
+	if m.XXX_unrecognized != nil {
+		i -= len(m.XXX_unrecognized)
+		copy(dAtA[i:], m.XXX_unrecognized)
+	}
+	if len(m.Stores) > 0 {
+		for k := range m.Stores {
+			v := m.Stores[k]
+			baseI := i
+			{
+				size, err := (&v).MarshalToSizedBuffer(dAtA[:i])
+				if err != nil {
+					return 0, err
+				}
+				i -= size
+				i = encodeVarintHakeeper(dAtA, i, uint64(size))
+			}
+			i--
+			dAtA[i] = 0x12
+			i -= len(k)
+			copy(dAtA[i:], k)
+			i = encodeVarintHakeeper(dAtA, i, uint64(len(k)))
+			i--
+			dAtA[i] = 0xa
+			i = encodeVarintHakeeper(dAtA, i, uint64(baseI-i))
+			i--
+			dAtA[i] = 0x12
+		}
+	}
+	if len(m.Shards) > 0 {
+		for k := range m.Shards {
+			v := m.Shards[k]
+			baseI := i
+			{
+				size, err := (&v).MarshalToSizedBuffer(dAtA[:i])
+				if err != nil {
+					return 0, err
+				}
+				i -= size
+				i = encodeVarintHakeeper(dAtA, i, uint64(size))
+			}
+			i--
+			dAtA[i] = 0x12
+			i = encodeVarintHakeeper(dAtA, i, uint64(k))
+			i--
+			dAtA[i] = 0x8
+			i = encodeVarintHakeeper(dAtA, i, uint64(baseI-i))
+			i--
+			dAtA[i] = 0xa
+		}
+	}
+	return len(dAtA) - i, nil
+}
+
+func (m *HAKeeperState) Marshal() (dAtA []byte, err error) {
+	size := m.Size()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	if err != nil {
+		return nil, err
+	}
+	return dAtA[:n], nil
+}
+
+func (m *HAKeeperState) MarshalTo(dAtA []byte) (int, error) {
+	size := m.Size()
+	return m.MarshalToSizedBuffer(dAtA[:size])
+}
+
+func (m *HAKeeperState) MarshalToSizedBuffer(dAtA []byte) (int, error) {
+	i := len(dAtA)
+	_ = i
+	var l int
+	_ = l
+	if m.XXX_unrecognized != nil {
+		i -= len(m.XXX_unrecognized)
+		copy(dAtA[i:], m.XXX_unrecognized)
+	}
+	{
+		size, err := m.LogState.MarshalToSizedBuffer(dAtA[:i])
+		if err != nil {
+			return 0, err
+		}
+		i -= size
+		i = encodeVarintHakeeper(dAtA, i, uint64(size))
+	}
+	i--
+	dAtA[i] = 0x22
+	{
+		size, err := m.DNState.MarshalToSizedBuffer(dAtA[:i])
+		if err != nil {
+			return 0, err
+		}
+		i -= size
+		i = encodeVarintHakeeper(dAtA, i, uint64(size))
+	}
+	i--
+	dAtA[i] = 0x1a
+	{
+		size, err := m.ClusterInfo.MarshalToSizedBuffer(dAtA[:i])
+		if err != nil {
+			return 0, err
+		}
+		i -= size
+		i = encodeVarintHakeeper(dAtA, i, uint64(size))
+	}
+	i--
+	dAtA[i] = 0x12
+	if m.Tick != 0 {
+		i = encodeVarintHakeeper(dAtA, i, uint64(m.Tick))
+		i--
+		dAtA[i] = 0x8
+	}
+	return len(dAtA) - i, nil
+}
+
+func (m *RSMState) Marshal() (dAtA []byte, err error) {
+	size := m.Size()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	if err != nil {
+		return nil, err
+	}
+	return dAtA[:n], nil
+}
+
+func (m *RSMState) MarshalTo(dAtA []byte) (int, error) {
+	size := m.Size()
+	return m.MarshalToSizedBuffer(dAtA[:size])
+}
+
+func (m *RSMState) MarshalToSizedBuffer(dAtA []byte) (int, error) {
+	i := len(dAtA)
+	_ = i
+	var l int
+	_ = l
+	if m.XXX_unrecognized != nil {
+		i -= len(m.XXX_unrecognized)
+		copy(dAtA[i:], m.XXX_unrecognized)
+	}
+	{
+		size, err := m.ClusterInfo.MarshalToSizedBuffer(dAtA[:i])
+		if err != nil {
+			return 0, err
+		}
+		i -= size
+		i = encodeVarintHakeeper(dAtA, i, uint64(size))
+	}
+	i--
+	dAtA[i] = 0x42
+	{
+		size, err := m.LogState.MarshalToSizedBuffer(dAtA[:i])
+		if err != nil {
+			return 0, err
+		}
+		i -= size
+		i = encodeVarintHakeeper(dAtA, i, uint64(size))
+	}
+	i--
+	dAtA[i] = 0x3a
+	{
+		size, err := m.DNState.MarshalToSizedBuffer(dAtA[:i])
+		if err != nil {
+			return 0, err
+		}
+		i -= size
+		i = encodeVarintHakeeper(dAtA, i, uint64(size))
+	}
+	i--
+	dAtA[i] = 0x32
+	if len(m.LogShards) > 0 {
+		for k := range m.LogShards {
+			v := m.LogShards[k]
+			baseI := i
+			i = encodeVarintHakeeper(dAtA, i, uint64(v))
+			i--
+			dAtA[i] = 0x10
+			i -= len(k)
+			copy(dAtA[i:], k)
+			i = encodeVarintHakeeper(dAtA, i, uint64(len(k)))
+			i--
+			dAtA[i] = 0xa
+			i = encodeVarintHakeeper(dAtA, i, uint64(baseI-i))
+			i--
+			dAtA[i] = 0x2a
+		}
+	}
+	if len(m.ScheduleCommands) > 0 {
+		for k := range m.ScheduleCommands {
+			v := m.ScheduleCommands[k]
+			baseI := i
+			{
+				size, err := (&v).MarshalToSizedBuffer(dAtA[:i])
+				if err != nil {
+					return 0, err
+				}
+				i -= size
+				i = encodeVarintHakeeper(dAtA, i, uint64(size))
+			}
+			i--
+			dAtA[i] = 0x12
+			i -= len(k)
+			copy(dAtA[i:], k)
+			i = encodeVarintHakeeper(dAtA, i, uint64(len(k)))
+			i--
+			dAtA[i] = 0xa
+			i = encodeVarintHakeeper(dAtA, i, uint64(baseI-i))
+			i--
+			dAtA[i] = 0x22
+		}
+	}
+	if m.Term != 0 {
+		i = encodeVarintHakeeper(dAtA, i, uint64(m.Term))
+		i--
+		dAtA[i] = 0x18
+	}
+	if m.NextID != 0 {
+		i = encodeVarintHakeeper(dAtA, i, uint64(m.NextID))
+		i--
+		dAtA[i] = 0x10
+	}
+	if m.Tick != 0 {
+		i = encodeVarintHakeeper(dAtA, i, uint64(m.Tick))
+		i--
+		dAtA[i] = 0x8
+	}
+	return len(dAtA) - i, nil
+}
+
+func encodeVarintHakeeper(dAtA []byte, offset int, v uint64) int {
+	offset -= sovHakeeper(v)
+	base := offset
+	for v >= 1<<7 {
+		dAtA[offset] = uint8(v&0x7f | 0x80)
+		v >>= 7
+		offset++
+	}
+	dAtA[offset] = uint8(v)
+	return base
+}
+func (m *Replica) Size() (n int) {
+	if m == nil {
+		return 0
+	}
+	var l int
+	_ = l
+	if m.ShardID != 0 {
+		n += 1 + sovHakeeper(uint64(m.ShardID))
+	}
+	if m.ReplicaID != 0 {
+		n += 1 + sovHakeeper(uint64(m.ReplicaID))
+	}
+	if m.Epoch != 0 {
+		n += 1 + sovHakeeper(uint64(m.Epoch))
+	}
+	if m.XXX_unrecognized != nil {
+		n += len(m.XXX_unrecognized)
+	}
+	return n
+}
+
+func (m *ConfigChange) Size() (n int) {
+	if m == nil {
+		return 0
+	}
+	var l int
+	_ = l
+	l = m.Replica.Size()
+	n += 1 + l + sovHakeeper(uint64(l))
+	if m.ChangeType != 0 {
+		n += 1 + sovHakeeper(uint64(m.ChangeType))
+	}
+	if m.XXX_unrecognized != nil {
+		n += len(m.XXX_unrecognized)
+	}
+	return n
+}
+
+func (m *ScheduleCommand) Size() (n int) {
+	if m == nil {
+		return 0
+	}
+	var l int
+	_ = l
+	l = len(m.UUID)
+	if l > 0 {
+		n += 1 + l + sovHakeeper(uint64(l))
+	}
+	l = m.ConfigChange.Size()
+	n += 1 + l + sovHakeeper(uint64(l))
+	if m.ServiceType != 0 {
+		n += 1 + sovHakeeper(uint64(m.ServiceType))
+	}
+	if m.XXX_unrecognized != nil {
+		n += len(m.XXX_unrecognized)
+	}
+	return n
+}
+
+func (m *CommandBatch) Size() (n int) {
+	if m == nil {
+		return 0
+	}
+	var l int
+	_ = l
+	if m.Term != 0 {
+		n += 1 + sovHakeeper(uint64(m.Term))
+	}
+	if len(m.Commands) > 0 {
+		for _, e := range m.Commands {
+			l = e.Size()
+			n += 1 + l + sovHakeeper(uint64(l))
+		}
+	}
+	if m.XXX_unrecognized != nil {
+		n += len(m.XXX_unrecognized)
+	}
+	return n
+}
+
+func (m *DNStoreInfo) Size() (n int) {
+	if m == nil {
+		return 0
+	}
+	var l int
+	_ = l
+	if m.Tick != 0 {
+		n += 1 + sovHakeeper(uint64(m.Tick))
+	}
+	if len(m.Shards) > 0 {
+		for _, e := range m.Shards {
+			l = e.Size()
+			n += 1 + l + sovHakeeper(uint64(l))
+		}
+	}
+	if m.XXX_unrecognized != nil {
+		n += len(m.XXX_unrecognized)
+	}
+	return n
+}
+
+func (m *DNState) Size() (n int) {
+	if m == nil {
+		return 0
+	}
+	var l int
+	_ = l
+	if len(m.Stores) > 0 {
+		for k, v := range m.Stores {
+			_ = k
+			_ = v
+			l = v.Size()
+			mapEntrySize := 1 + len(k) + sovHakeeper(uint64(len(k))) + 1 + l + sovHakeeper(uint64(l))
+			n += mapEntrySize + 1 + sovHakeeper(uint64(mapEntrySize))
+		}
+	}
+	if m.XXX_unrecognized != nil {
+		n += len(m.XXX_unrecognized)
+	}
+	return n
+}
+
+func (m *ClusterInfo) Size() (n int) {
+	if m == nil {
+		return 0
+	}
+	var l int
+	_ = l
+	if len(m.DNShards) > 0 {
+		for _, e := range m.DNShards {
+			l = e.Size()
+			n += 1 + l + sovHakeeper(uint64(l))
+		}
+	}
+	if len(m.LogShards) > 0 {
+		for _, e := range m.LogShards {
+			l = e.Size()
+			n += 1 + l + sovHakeeper(uint64(l))
+		}
+	}
+	if m.XXX_unrecognized != nil {
+		n += len(m.XXX_unrecognized)
+	}
+	return n
+}
+
+func (m *LogStoreInfo) Size() (n int) {
+	if m == nil {
+		return 0
+	}
+	var l int
+	_ = l
+	if m.Tick != 0 {
+		n += 1 + sovHakeeper(uint64(m.Tick))
+	}
+	l = len(m.RaftAddress)
+	if l > 0 {
+		n += 1 + l + sovHakeeper(uint64(l))
+	}
+	l = len(m.ServiceAddress)
+	if l > 0 {
+		n += 1 + l + sovHakeeper(uint64(l))
+	}
+	l = len(m.GossipAddress)
+	if l > 0 {
+		n += 1 + l + sovHakeeper(uint64(l))
+	}
+	if len(m.Replicas) > 0 {
+		for _, e := range m.Replicas {
+			l = e.Size()
+			n += 1 + l + sovHakeeper(uint64(l))
+		}
+	}
+	if m.XXX_unrecognized != nil {
+		n += len(m.XXX_unrecognized)
+	}
+	return n
+}
+
+func (m *LogState) Size() (n int) {
+	if m == nil {
+		return 0
+	}
+	var l int
+	_ = l
+	if len(m.Shards) > 0 {
+		for k, v := range m.Shards {
+			_ = k
+			_ = v
+			l = v.Size()
+			mapEntrySize := 1 + sovHakeeper(uint64(k)) + 1 + l + sovHakeeper(uint64(l))
+			n += mapEntrySize + 1 + sovHakeeper(uint64(mapEntrySize))
+		}
+	}
+	if len(m.Stores) > 0 {
+		for k, v := range m.Stores {
+			_ = k
+			_ = v
+			l = v.Size()
+			mapEntrySize := 1 + len(k) + sovHakeeper(uint64(len(k))) + 1 + l + sovHakeeper(uint64(l))
+			n += mapEntrySize + 1 + sovHakeeper(uint64(mapEntrySize))
+		}
+	}
+	if m.XXX_unrecognized != nil {
+		n += len(m.XXX_unrecognized)
+	}
+	return n
+}
+
+func (m *HAKeeperState) Size() (n int) {
+	if m == nil {
+		return 0
+	}
+	var l int
+	_ = l
+	if m.Tick != 0 {
+		n += 1 + sovHakeeper(uint64(m.Tick))
+	}
+	l = m.ClusterInfo.Size()
+	n += 1 + l + sovHakeeper(uint64(l))
+	l = m.DNState.Size()
+	n += 1 + l + sovHakeeper(uint64(l))
+	l = m.LogState.Size()
+	n += 1 + l + sovHakeeper(uint64(l))
+	if m.XXX_unrecognized != nil {
+		n += len(m.XXX_unrecognized)
+	}
+	return n
+}
+
+func (m *RSMState) Size() (n int) {
+	if m == nil {
+		return 0
+	}
+	var l int
+	_ = l
+	if m.Tick != 0 {
+		n += 1 + sovHakeeper(uint64(m.Tick))
+	}
+	if m.NextID != 0 {
+		n += 1 + sovHakeeper(uint64(m.NextID))
+	}
+	if m.Term != 0 {
+		n += 1 + sovHakeeper(uint64(m.Term))
+	}
+	if len(m.ScheduleCommands) > 0 {
+		for k, v := range m.ScheduleCommands {
+			_ = k
+			_ = v
+			l = v.Size()
+			mapEntrySize := 1 + len(k) + sovHakeeper(uint64(len(k))) + 1 + l + sovHakeeper(uint64(l))
+			n += mapEntrySize + 1 + sovHakeeper(uint64(mapEntrySize))
+		}
+	}
+	if len(m.LogShards) > 0 {
+		for k, v := range m.LogShards {
+			_ = k
+			_ = v
+			mapEntrySize := 1 + len(k) + sovHakeeper(uint64(len(k))) + 1 + sovHakeeper(uint64(v))
+			n += mapEntrySize + 1 + sovHakeeper(uint64(mapEntrySize))
+		}
+	}
+	l = m.DNState.Size()
+	n += 1 + l + sovHakeeper(uint64(l))
+	l = m.LogState.Size()
+	n += 1 + l + sovHakeeper(uint64(l))
+	l = m.ClusterInfo.Size()
+	n += 1 + l + sovHakeeper(uint64(l))
+	if m.XXX_unrecognized != nil {
+		n += len(m.XXX_unrecognized)
+	}
+	return n
+}
+
+func sovHakeeper(x uint64) (n int) {
+	return (math_bits.Len64(x|1) + 6) / 7
+}
+func sozHakeeper(x uint64) (n int) {
+	return sovHakeeper(uint64((x << 1) ^ uint64((int64(x) >> 63))))
+}
+func (m *Replica) Unmarshal(dAtA []byte) error {
+	l := len(dAtA)
+	iNdEx := 0
+	for iNdEx < l {
+		preIndex := iNdEx
+		var wire uint64
+		for shift := uint(0); ; shift += 7 {
+			if shift >= 64 {
+				return ErrIntOverflowHakeeper
+			}
+			if iNdEx >= l {
+				return io.ErrUnexpectedEOF
+			}
+			b := dAtA[iNdEx]
+			iNdEx++
+			wire |= uint64(b&0x7F) << shift
+			if b < 0x80 {
+				break
+			}
+		}
+		fieldNum := int32(wire >> 3)
+		wireType := int(wire & 0x7)
+		if wireType == 4 {
+			return fmt.Errorf("proto: Replica: wiretype end group for non-group")
+		}
+		if fieldNum <= 0 {
+			return fmt.Errorf("proto: Replica: illegal tag %d (wire type %d)", fieldNum, wire)
+		}
+		switch fieldNum {
+		case 1:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field ShardID", wireType)
+			}
+			m.ShardID = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowHakeeper
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.ShardID |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+		case 2:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field ReplicaID", wireType)
+			}
+			m.ReplicaID = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowHakeeper
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.ReplicaID |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+		case 3:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Epoch", wireType)
+			}
+			m.Epoch = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowHakeeper
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.Epoch |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+		default:
+			iNdEx = preIndex
+			skippy, err := skipHakeeper(dAtA[iNdEx:])
+			if err != nil {
+				return err
+			}
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			if (iNdEx + skippy) > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
+			iNdEx += skippy
+		}
+	}
+
+	if iNdEx > l {
+		return io.ErrUnexpectedEOF
+	}
+	return nil
+}
+func (m *ConfigChange) Unmarshal(dAtA []byte) error {
+	l := len(dAtA)
+	iNdEx := 0
+	for iNdEx < l {
+		preIndex := iNdEx
+		var wire uint64
+		for shift := uint(0); ; shift += 7 {
+			if shift >= 64 {
+				return ErrIntOverflowHakeeper
+			}
+			if iNdEx >= l {
+				return io.ErrUnexpectedEOF
+			}
+			b := dAtA[iNdEx]
+			iNdEx++
+			wire |= uint64(b&0x7F) << shift
+			if b < 0x80 {
+				break
+			}
+		}
+		fieldNum := int32(wire >> 3)
+		wireType := int(wire & 0x7)
+		if wireType == 4 {
+			return fmt.Errorf("proto: ConfigChange: wiretype end group for non-group")
+		}
+		if fieldNum <= 0 {
+			return fmt.Errorf("proto: ConfigChange: illegal tag %d (wire type %d)", fieldNum, wire)
+		}
+		switch fieldNum {
+		case 1:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Replica", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowHakeeper
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			postIndex := iNdEx + msglen
+			if postIndex < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			if err := m.Replica.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
+			iNdEx = postIndex
+		case 2:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field ChangeType", wireType)
+			}
+			m.ChangeType = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowHakeeper
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.ChangeType |= ConfigChangeType(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+		default:
+			iNdEx = preIndex
+			skippy, err := skipHakeeper(dAtA[iNdEx:])
+			if err != nil {
+				return err
+			}
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			if (iNdEx + skippy) > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
+			iNdEx += skippy
+		}
+	}
+
+	if iNdEx > l {
+		return io.ErrUnexpectedEOF
+	}
+	return nil
+}
+func (m *ScheduleCommand) Unmarshal(dAtA []byte) error {
+	l := len(dAtA)
+	iNdEx := 0
+	for iNdEx < l {
+		preIndex := iNdEx
+		var wire uint64
+		for shift := uint(0); ; shift += 7 {
+			if shift >= 64 {
+				return ErrIntOverflowHakeeper
+			}
+			if iNdEx >= l {
+				return io.ErrUnexpectedEOF
+			}
+			b := dAtA[iNdEx]
+			iNdEx++
+			wire |= uint64(b&0x7F) << shift
+			if b < 0x80 {
+				break
+			}
+		}
+		fieldNum := int32(wire >> 3)
+		wireType := int(wire & 0x7)
+		if wireType == 4 {
+			return fmt.Errorf("proto: ScheduleCommand: wiretype end group for non-group")
+		}
+		if fieldNum <= 0 {
+			return fmt.Errorf("proto: ScheduleCommand: illegal tag %d (wire type %d)", fieldNum, wire)
+		}
+		switch fieldNum {
+		case 1:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field UUID", wireType)
+			}
+			var stringLen uint64
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowHakeeper
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				stringLen |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			intStringLen := int(stringLen)
+			if intStringLen < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			postIndex := iNdEx + intStringLen
+			if postIndex < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.UUID = string(dAtA[iNdEx:postIndex])
+			iNdEx = postIndex
+		case 2:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field ConfigChange", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowHakeeper
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			postIndex := iNdEx + msglen
+			if postIndex < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			if err := m.ConfigChange.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
+			iNdEx = postIndex
+		case 3:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field ServiceType", wireType)
+			}
+			m.ServiceType = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowHakeeper
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.ServiceType |= ServiceType(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+		default:
+			iNdEx = preIndex
+			skippy, err := skipHakeeper(dAtA[iNdEx:])
+			if err != nil {
+				return err
+			}
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			if (iNdEx + skippy) > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
+			iNdEx += skippy
+		}
+	}
+
+	if iNdEx > l {
+		return io.ErrUnexpectedEOF
+	}
+	return nil
+}
+func (m *CommandBatch) Unmarshal(dAtA []byte) error {
+	l := len(dAtA)
+	iNdEx := 0
+	for iNdEx < l {
+		preIndex := iNdEx
+		var wire uint64
+		for shift := uint(0); ; shift += 7 {
+			if shift >= 64 {
+				return ErrIntOverflowHakeeper
+			}
+			if iNdEx >= l {
+				return io.ErrUnexpectedEOF
+			}
+			b := dAtA[iNdEx]
+			iNdEx++
+			wire |= uint64(b&0x7F) << shift
+			if b < 0x80 {
+				break
+			}
+		}
+		fieldNum := int32(wire >> 3)
+		wireType := int(wire & 0x7)
+		if wireType == 4 {
+			return fmt.Errorf("proto: CommandBatch: wiretype end group for non-group")
+		}
+		if fieldNum <= 0 {
+			return fmt.Errorf("proto: CommandBatch: illegal tag %d (wire type %d)", fieldNum, wire)
+		}
+		switch fieldNum {
+		case 1:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Term", wireType)
+			}
+			m.Term = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowHakeeper
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.Term |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+		case 2:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Commands", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowHakeeper
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			postIndex := iNdEx + msglen
+			if postIndex < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.Commands = append(m.Commands, ScheduleCommand{})
+			if err := m.Commands[len(m.Commands)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
+			iNdEx = postIndex
+		default:
+			iNdEx = preIndex
+			skippy, err := skipHakeeper(dAtA[iNdEx:])
+			if err != nil {
+				return err
+			}
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			if (iNdEx + skippy) > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
+			iNdEx += skippy
+		}
+	}
+
+	if iNdEx > l {
+		return io.ErrUnexpectedEOF
+	}
+	return nil
+}
+func (m *DNStoreInfo) Unmarshal(dAtA []byte) error {
+	l := len(dAtA)
+	iNdEx := 0
+	for iNdEx < l {
+		preIndex := iNdEx
+		var wire uint64
+		for shift := uint(0); ; shift += 7 {
+			if shift >= 64 {
+				return ErrIntOverflowHakeeper
+			}
+			if iNdEx >= l {
+				return io.ErrUnexpectedEOF
+			}
+			b := dAtA[iNdEx]
+			iNdEx++
+			wire |= uint64(b&0x7F) << shift
+			if b < 0x80 {
+				break
+			}
+		}
+		fieldNum := int32(wire >> 3)
+		wireType := int(wire & 0x7)
+		if wireType == 4 {
+			return fmt.Errorf("proto: DNStoreInfo: wiretype end group for non-group")
+		}
+		if fieldNum <= 0 {
+			return fmt.Errorf("proto: DNStoreInfo: illegal tag %d (wire type %d)", fieldNum, wire)
+		}
+		switch fieldNum {
+		case 1:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Tick", wireType)
+			}
+			m.Tick = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowHakeeper
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.Tick |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+		case 2:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Shards", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowHakeeper
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			postIndex := iNdEx + msglen
+			if postIndex < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.Shards = append(m.Shards, logservice.DNShardInfo{})
+			if err := m.Shards[len(m.Shards)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
+			iNdEx = postIndex
+		default:
+			iNdEx = preIndex
+			skippy, err := skipHakeeper(dAtA[iNdEx:])
+			if err != nil {
+				return err
+			}
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			if (iNdEx + skippy) > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
+			iNdEx += skippy
+		}
+	}
+
+	if iNdEx > l {
+		return io.ErrUnexpectedEOF
+	}
+	return nil
+}
+func (m *DNState) Unmarshal(dAtA []byte) error {
+	l := len(dAtA)
+	iNdEx := 0
+	for iNdEx < l {
+		preIndex := iNdEx
+		var wire uint64
+		for shift := uint(0); ; shift += 7 {
+			if shift >= 64 {
+				return ErrIntOverflowHakeeper
+			}
+			if iNdEx >= l {
+				return io.ErrUnexpectedEOF
+			}
+			b := dAtA[iNdEx]
+			iNdEx++
+			wire |= uint64(b&0x7F) << shift
+			if b < 0x80 {
+				break
+			}
+		}
+		fieldNum := int32(wire >> 3)
+		wireType := int(wire & 0x7)
+		if wireType == 4 {
+			return fmt.Errorf("proto: DNState: wiretype end group for non-group")
+		}
+		if fieldNum <= 0 {
+			return fmt.Errorf("proto: DNState: illegal tag %d (wire type %d)", fieldNum, wire)
+		}
+		switch fieldNum {
+		case 1:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Stores", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowHakeeper
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			postIndex := iNdEx + msglen
+			if postIndex < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			if m.Stores == nil {
+				m.Stores = make(map[string]DNStoreInfo)
+			}
+			var mapkey string
+			mapvalue := &DNStoreInfo{}
+			for iNdEx < postIndex {
+				entryPreIndex := iNdEx
+				var wire uint64
+				for shift := uint(0); ; shift += 7 {
+					if shift >= 64 {
+						return ErrIntOverflowHakeeper
+					}
+					if iNdEx >= l {
+						return io.ErrUnexpectedEOF
+					}
+					b := dAtA[iNdEx]
+					iNdEx++
+					wire |= uint64(b&0x7F) << shift
+					if b < 0x80 {
+						break
+					}
+				}
+				fieldNum := int32(wire >> 3)
+				if fieldNum == 1 {
+					var stringLenmapkey uint64
+					for shift := uint(0); ; shift += 7 {
+						if shift >= 64 {
+							return ErrIntOverflowHakeeper
+						}
+						if iNdEx >= l {
+							return io.ErrUnexpectedEOF
+						}
+						b := dAtA[iNdEx]
+						iNdEx++
+						stringLenmapkey |= uint64(b&0x7F) << shift
+						if b < 0x80 {
+							break
+						}
+					}
+					intStringLenmapkey := int(stringLenmapkey)
+					if intStringLenmapkey < 0 {
+						return ErrInvalidLengthHakeeper
+					}
+					postStringIndexmapkey := iNdEx + intStringLenmapkey
+					if postStringIndexmapkey < 0 {
+						return ErrInvalidLengthHakeeper
+					}
+					if postStringIndexmapkey > l {
+						return io.ErrUnexpectedEOF
+					}
+					mapkey = string(dAtA[iNdEx:postStringIndexmapkey])
+					iNdEx = postStringIndexmapkey
+				} else if fieldNum == 2 {
+					var mapmsglen int
+					for shift := uint(0); ; shift += 7 {
+						if shift >= 64 {
+							return ErrIntOverflowHakeeper
+						}
+						if iNdEx >= l {
+							return io.ErrUnexpectedEOF
+						}
+						b := dAtA[iNdEx]
+						iNdEx++
+						mapmsglen |= int(b&0x7F) << shift
+						if b < 0x80 {
+							break
+						}
+					}
+					if mapmsglen < 0 {
+						return ErrInvalidLengthHakeeper
+					}
+					postmsgIndex := iNdEx + mapmsglen
+					if postmsgIndex < 0 {
+						return ErrInvalidLengthHakeeper
+					}
+					if postmsgIndex > l {
+						return io.ErrUnexpectedEOF
+					}
+					mapvalue = &DNStoreInfo{}
+					if err := mapvalue.Unmarshal(dAtA[iNdEx:postmsgIndex]); err != nil {
+						return err
+					}
+					iNdEx = postmsgIndex
+				} else {
+					iNdEx = entryPreIndex
+					skippy, err := skipHakeeper(dAtA[iNdEx:])
+					if err != nil {
+						return err
+					}
+					if (skippy < 0) || (iNdEx+skippy) < 0 {
+						return ErrInvalidLengthHakeeper
+					}
+					if (iNdEx + skippy) > postIndex {
+						return io.ErrUnexpectedEOF
+					}
+					iNdEx += skippy
+				}
+			}
+			m.Stores[mapkey] = *mapvalue
+			iNdEx = postIndex
+		default:
+			iNdEx = preIndex
+			skippy, err := skipHakeeper(dAtA[iNdEx:])
+			if err != nil {
+				return err
+			}
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			if (iNdEx + skippy) > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
+			iNdEx += skippy
+		}
 	}
-	return n
-}
 
-func (m *ScheduleCommand) Size() (n int) {
-	if m == nil {
-		return 0
-	}
-	var l int
-	_ = l
-	l = len(m.UUID)
-	if l > 0 {
-		n += 1 + l + sovHakeeper(uint64(l))
-	}
-	l = m.ConfigChange.Size()
-	n += 1 + l + sovHakeeper(uint64(l))
-	if m.ServiceType != 0 {
-		n += 1 + sovHakeeper(uint64(m.ServiceType))
-	}
-	if m.XXX_unrecognized != nil {
-		n += len(m.XXX_unrecognized)
+	if iNdEx > l {
+		return io.ErrUnexpectedEOF
 	}
-	return n
+	return nil
 }
-
-func (m *CommandBatch) Size() (n int) {
-	if m == nil {
-		return 0
-	}
-	var l int
-	_ = l
-	if m.Term != 0 {
-		n += 1 + sovHakeeper(uint64(m.Term))
-	}
-	if len(m.Commands) > 0 {
-		for _, e := range m.Commands {
-			l = e.Size()
-			n += 1 + l + sovHakeeper(uint64(l))
+func (m *ClusterInfo) Unmarshal(dAtA []byte) error {
+	l := len(dAtA)
+	iNdEx := 0
+	for iNdEx < l {
+		preIndex := iNdEx
+		var wire uint64
+		for shift := uint(0); ; shift += 7 {
+			if shift >= 64 {
+				return ErrIntOverflowHakeeper
+			}
+			if iNdEx >= l {
+				return io.ErrUnexpectedEOF
+			}
+			b := dAtA[iNdEx]
+			iNdEx++
+			wire |= uint64(b&0x7F) << shift
+			if b < 0x80 {
+				break
+			}
+		}
+		fieldNum := int32(wire >> 3)
+		wireType := int(wire & 0x7)
+		if wireType == 4 {
+			return fmt.Errorf("proto: ClusterInfo: wiretype end group for non-group")
+		}
+		if fieldNum <= 0 {
+			return fmt.Errorf("proto: ClusterInfo: illegal tag %d (wire type %d)", fieldNum, wire)
+		}
+		switch fieldNum {
+		case 1:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field DNShards", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowHakeeper
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			postIndex := iNdEx + msglen
+			if postIndex < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.DNShards = append(m.DNShards, metadata.DNShardRecord{})
+			if err := m.DNShards[len(m.DNShards)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
+			iNdEx = postIndex
+		case 2:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field LogShards", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowHakeeper
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			postIndex := iNdEx + msglen
+			if postIndex < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.LogShards = append(m.LogShards, metadata.LogShardRecord{})
+			if err := m.LogShards[len(m.LogShards)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
+			iNdEx = postIndex
+		default:
+			iNdEx = preIndex
+			skippy, err := skipHakeeper(dAtA[iNdEx:])
+			if err != nil {
+				return err
+			}
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			if (iNdEx + skippy) > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
+			iNdEx += skippy
 		}
 	}
-	if m.XXX_unrecognized != nil {
-		n += len(m.XXX_unrecognized)
-	}
-	return n
-}
 
-func sovHakeeper(x uint64) (n int) {
-	return (math_bits.Len64(x|1) + 6) / 7
-}
-func sozHakeeper(x uint64) (n int) {
-	return sovHakeeper(uint64((x << 1) ^ uint64((int64(x) >> 63))))
+	if iNdEx > l {
+		return io.ErrUnexpectedEOF
+	}
+	return nil
 }
-func (m *Replica) Unmarshal(dAtA []byte) error {
+func (m *LogStoreInfo) Unmarshal(dAtA []byte) error {
 	l := len(dAtA)
 	iNdEx := 0
 	for iNdEx < l {
@@ -659,17 +2700,17 @@ func (m *Replica) Unmarshal(dAtA []byte) error {
 		fieldNum := int32(wire >> 3)
 		wireType := int(wire & 0x7)
 		if wireType == 4 {
-			return fmt.Errorf("proto: Replica: wiretype end group for non-group")
+			return fmt.Errorf("proto: LogStoreInfo: wiretype end group for non-group")
 		}
 		if fieldNum <= 0 {
-			return fmt.Errorf("proto: Replica: illegal tag %d (wire type %d)", fieldNum, wire)
+			return fmt.Errorf("proto: LogStoreInfo: illegal tag %d (wire type %d)", fieldNum, wire)
 		}
 		switch fieldNum {
 		case 1:
 			if wireType != 0 {
-				return fmt.Errorf("proto: wrong wireType = %d for field ShardID", wireType)
+				return fmt.Errorf("proto: wrong wireType = %d for field Tick", wireType)
 			}
-			m.ShardID = 0
+			m.Tick = 0
 			for shift := uint(0); ; shift += 7 {
 				if shift >= 64 {
 					return ErrIntOverflowHakeeper
@@ -679,16 +2720,16 @@ func (m *Replica) Unmarshal(dAtA []byte) error {
 				}
 				b := dAtA[iNdEx]
 				iNdEx++
-				m.ShardID |= uint64(b&0x7F) << shift
+				m.Tick |= uint64(b&0x7F) << shift
 				if b < 0x80 {
 					break
 				}
 			}
 		case 2:
-			if wireType != 0 {
-				return fmt.Errorf("proto: wrong wireType = %d for field ReplicaID", wireType)
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field RaftAddress", wireType)
 			}
-			m.ReplicaID = 0
+			var stringLen uint64
 			for shift := uint(0); ; shift += 7 {
 				if shift >= 64 {
 					return ErrIntOverflowHakeeper
@@ -698,16 +2739,29 @@ func (m *Replica) Unmarshal(dAtA []byte) error {
 				}
 				b := dAtA[iNdEx]
 				iNdEx++
-				m.ReplicaID |= uint64(b&0x7F) << shift
+				stringLen |= uint64(b&0x7F) << shift
 				if b < 0x80 {
 					break
 				}
 			}
+			intStringLen := int(stringLen)
+			if intStringLen < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			postIndex := iNdEx + intStringLen
+			if postIndex < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.RaftAddress = string(dAtA[iNdEx:postIndex])
+			iNdEx = postIndex
 		case 3:
-			if wireType != 0 {
-				return fmt.Errorf("proto: wrong wireType = %d for field Epoch", wireType)
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field ServiceAddress", wireType)
 			}
-			m.Epoch = 0
+			var stringLen uint64
 			for shift := uint(0); ; shift += 7 {
 				if shift >= 64 {
 					return ErrIntOverflowHakeeper
@@ -717,11 +2771,90 @@ func (m *Replica) Unmarshal(dAtA []byte) error {
 				}
 				b := dAtA[iNdEx]
 				iNdEx++
-				m.Epoch |= uint64(b&0x7F) << shift
+				stringLen |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			intStringLen := int(stringLen)
+			if intStringLen < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			postIndex := iNdEx + intStringLen
+			if postIndex < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.ServiceAddress = string(dAtA[iNdEx:postIndex])
+			iNdEx = postIndex
+		case 4:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field GossipAddress", wireType)
+			}
+			var stringLen uint64
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowHakeeper
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				stringLen |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			intStringLen := int(stringLen)
+			if intStringLen < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			postIndex := iNdEx + intStringLen
+			if postIndex < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.GossipAddress = string(dAtA[iNdEx:postIndex])
+			iNdEx = postIndex
+		case 5:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Replicas", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowHakeeper
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= int(b&0x7F) << shift
 				if b < 0x80 {
 					break
 				}
 			}
+			if msglen < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			postIndex := iNdEx + msglen
+			if postIndex < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.Replicas = append(m.Replicas, logservice.LogReplicaInfo{})
+			if err := m.Replicas[len(m.Replicas)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
+			iNdEx = postIndex
 		default:
 			iNdEx = preIndex
 			skippy, err := skipHakeeper(dAtA[iNdEx:])
@@ -744,7 +2877,7 @@ func (m *Replica) Unmarshal(dAtA []byte) error {
 	}
 	return nil
 }
-func (m *ConfigChange) Unmarshal(dAtA []byte) error {
+func (m *LogState) Unmarshal(dAtA []byte) error {
 	l := len(dAtA)
 	iNdEx := 0
 	for iNdEx < l {
@@ -767,15 +2900,15 @@ func (m *ConfigChange) Unmarshal(dAtA []byte) error {
 		fieldNum := int32(wire >> 3)
 		wireType := int(wire & 0x7)
 		if wireType == 4 {
-			return fmt.Errorf("proto: ConfigChange: wiretype end group for non-group")
+			return fmt.Errorf("proto: LogState: wiretype end group for non-group")
 		}
 		if fieldNum <= 0 {
-			return fmt.Errorf("proto: ConfigChange: illegal tag %d (wire type %d)", fieldNum, wire)
+			return fmt.Errorf("proto: LogState: illegal tag %d (wire type %d)", fieldNum, wire)
 		}
 		switch fieldNum {
 		case 1:
 			if wireType != 2 {
-				return fmt.Errorf("proto: wrong wireType = %d for field Replica", wireType)
+				return fmt.Errorf("proto: wrong wireType = %d for field Shards", wireType)
 			}
 			var msglen int
 			for shift := uint(0); ; shift += 7 {
@@ -802,15 +2935,97 @@ func (m *ConfigChange) Unmarshal(dAtA []byte) error {
 			if postIndex > l {
 				return io.ErrUnexpectedEOF
 			}
-			if err := m.Replica.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
-				return err
+			if m.Shards == nil {
+				m.Shards = make(map[uint64]logservice.LogShardInfo)
+			}
+			var mapkey uint64
+			mapvalue := &logservice.LogShardInfo{}
+			for iNdEx < postIndex {
+				entryPreIndex := iNdEx
+				var wire uint64
+				for shift := uint(0); ; shift += 7 {
+					if shift >= 64 {
+						return ErrIntOverflowHakeeper
+					}
+					if iNdEx >= l {
+						return io.ErrUnexpectedEOF
+					}
+					b := dAtA[iNdEx]
+					iNdEx++
+					wire |= uint64(b&0x7F) << shift
+					if b < 0x80 {
+						break
+					}
+				}
+				fieldNum := int32(wire >> 3)
+				if fieldNum == 1 {
+					for shift := uint(0); ; shift += 7 {
+						if shift >= 64 {
+							return ErrIntOverflowHakeeper
+						}
+						if iNdEx >= l {
+							return io.ErrUnexpectedEOF
+						}
+						b := dAtA[iNdEx]
+						iNdEx++
+						mapkey |= uint64(b&0x7F) << shift
+						if b < 0x80 {
+							break
+						}
+					}
+				} else if fieldNum == 2 {
+					var mapmsglen int
+					for shift := uint(0); ; shift += 7 {
+						if shift >= 64 {
+							return ErrIntOverflowHakeeper
+						}
+						if iNdEx >= l {
+							return io.ErrUnexpectedEOF
+						}
+						b := dAtA[iNdEx]
+						iNdEx++
+						mapmsglen |= int(b&0x7F) << shift
+						if b < 0x80 {
+							break
+						}
+					}
+					if mapmsglen < 0 {
+						return ErrInvalidLengthHakeeper
+					}
+					postmsgIndex := iNdEx + mapmsglen
+					if postmsgIndex < 0 {
+						return ErrInvalidLengthHakeeper
+					}
+					if postmsgIndex > l {
+						return io.ErrUnexpectedEOF
+					}
+					mapvalue = &logservice.LogShardInfo{}
+					if err := mapvalue.Unmarshal(dAtA[iNdEx:postmsgIndex]); err != nil {
+						return err
+					}
+					iNdEx = postmsgIndex
+				} else {
+					iNdEx = entryPreIndex
+					skippy, err := skipHakeeper(dAtA[iNdEx:])
+					if err != nil {
+						return err
+					}
+					if (skippy < 0) || (iNdEx+skippy) < 0 {
+						return ErrInvalidLengthHakeeper
+					}
+					if (iNdEx + skippy) > postIndex {
+						return io.ErrUnexpectedEOF
+					}
+					iNdEx += skippy
+				}
 			}
+			m.Shards[mapkey] = *mapvalue
 			iNdEx = postIndex
 		case 2:
-			if wireType != 0 {
-				return fmt.Errorf("proto: wrong wireType = %d for field ChangeType", wireType)
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Stores", wireType)
 			}
-			m.ChangeType = 0
+			var msglen int
 			for shift := uint(0); ; shift += 7 {
 				if shift >= 64 {
 					return ErrIntOverflowHakeeper
@@ -820,11 +3035,121 @@ func (m *ConfigChange) Unmarshal(dAtA []byte) error {
 				}
 				b := dAtA[iNdEx]
 				iNdEx++
-				m.ChangeType |= ConfigChangeType(b&0x7F) << shift
+				msglen |= int(b&0x7F) << shift
 				if b < 0x80 {
 					break
 				}
 			}
+			if msglen < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			postIndex := iNdEx + msglen
+			if postIndex < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			if m.Stores == nil {
+				m.Stores = make(map[string]LogStoreInfo)
+			}
+			var mapkey string
+			mapvalue := &LogStoreInfo{}
+			for iNdEx < postIndex {
+				entryPreIndex := iNdEx
+				var wire uint64
+				for shift := uint(0); ; shift += 7 {
+					if shift >= 64 {
+						return ErrIntOverflowHakeeper
+					}
+					if iNdEx >= l {
+						return io.ErrUnexpectedEOF
+					}
+					b := dAtA[iNdEx]
+					iNdEx++
+					wire |= uint64(b&0x7F) << shift
+					if b < 0x80 {
+						break
+					}
+				}
+				fieldNum := int32(wire >> 3)
+				if fieldNum == 1 {
+					var stringLenmapkey uint64
+					for shift := uint(0); ; shift += 7 {
+						if shift >= 64 {
+							return ErrIntOverflowHakeeper
+						}
+						if iNdEx >= l {
+							return io.ErrUnexpectedEOF
+						}
+						b := dAtA[iNdEx]
+						iNdEx++
+						stringLenmapkey |= uint64(b&0x7F) << shift
+						if b < 0x80 {
+							break
+						}
+					}
+					intStringLenmapkey := int(stringLenmapkey)
+					if intStringLenmapkey < 0 {
+						return ErrInvalidLengthHakeeper
+					}
+					postStringIndexmapkey := iNdEx + intStringLenmapkey
+					if postStringIndexmapkey < 0 {
+						return ErrInvalidLengthHakeeper
+					}
+					if postStringIndexmapkey > l {
+						return io.ErrUnexpectedEOF
+					}
+					mapkey = string(dAtA[iNdEx:postStringIndexmapkey])
+					iNdEx = postStringIndexmapkey
+				} else if fieldNum == 2 {
+					var mapmsglen int
+					for shift := uint(0); ; shift += 7 {
+						if shift >= 64 {
+							return ErrIntOverflowHakeeper
+						}
+						if iNdEx >= l {
+							return io.ErrUnexpectedEOF
+						}
+						b := dAtA[iNdEx]
+						iNdEx++
+						mapmsglen |= int(b&0x7F) << shift
+						if b < 0x80 {
+							break
+						}
+					}
+					if mapmsglen < 0 {
+						return ErrInvalidLengthHakeeper
+					}
+					postmsgIndex := iNdEx + mapmsglen
+					if postmsgIndex < 0 {
+						return ErrInvalidLengthHakeeper
+					}
+					if postmsgIndex > l {
+						return io.ErrUnexpectedEOF
+					}
+					mapvalue = &LogStoreInfo{}
+					if err := mapvalue.Unmarshal(dAtA[iNdEx:postmsgIndex]); err != nil {
+						return err
+					}
+					iNdEx = postmsgIndex
+				} else {
+					iNdEx = entryPreIndex
+					skippy, err := skipHakeeper(dAtA[iNdEx:])
+					if err != nil {
+						return err
+					}
+					if (skippy < 0) || (iNdEx+skippy) < 0 {
+						return ErrInvalidLengthHakeeper
+					}
+					if (iNdEx + skippy) > postIndex {
+						return io.ErrUnexpectedEOF
+					}
+					iNdEx += skippy
+				}
+			}
+			m.Stores[mapkey] = *mapvalue
+			iNdEx = postIndex
 		default:
 			iNdEx = preIndex
 			skippy, err := skipHakeeper(dAtA[iNdEx:])
@@ -847,7 +3172,7 @@ func (m *ConfigChange) Unmarshal(dAtA []byte) error {
 	}
 	return nil
 }
-func (m *ScheduleCommand) Unmarshal(dAtA []byte) error {
+func (m *HAKeeperState) Unmarshal(dAtA []byte) error {
 	l := len(dAtA)
 	iNdEx := 0
 	for iNdEx < l {
@@ -870,17 +3195,36 @@ func (m *ScheduleCommand) Unmarshal(dAtA []byte) error {
 		fieldNum := int32(wire >> 3)
 		wireType := int(wire & 0x7)
 		if wireType == 4 {
-			return fmt.Errorf("proto: ScheduleCommand: wiretype end group for non-group")
+			return fmt.Errorf("proto: HAKeeperState: wiretype end group for non-group")
 		}
 		if fieldNum <= 0 {
-			return fmt.Errorf("proto: ScheduleCommand: illegal tag %d (wire type %d)", fieldNum, wire)
+			return fmt.Errorf("proto: HAKeeperState: illegal tag %d (wire type %d)", fieldNum, wire)
 		}
 		switch fieldNum {
 		case 1:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Tick", wireType)
+			}
+			m.Tick = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowHakeeper
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.Tick |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+		case 2:
 			if wireType != 2 {
-				return fmt.Errorf("proto: wrong wireType = %d for field UUID", wireType)
+				return fmt.Errorf("proto: wrong wireType = %d for field ClusterInfo", wireType)
 			}
-			var stringLen uint64
+			var msglen int
 			for shift := uint(0); ; shift += 7 {
 				if shift >= 64 {
 					return ErrIntOverflowHakeeper
@@ -890,27 +3234,28 @@ func (m *ScheduleCommand) Unmarshal(dAtA []byte) error {
 				}
 				b := dAtA[iNdEx]
 				iNdEx++
-				stringLen |= uint64(b&0x7F) << shift
+				msglen |= int(b&0x7F) << shift
 				if b < 0x80 {
 					break
 				}
 			}
-			intStringLen := int(stringLen)
-			if intStringLen < 0 {
+			if msglen < 0 {
 				return ErrInvalidLengthHakeeper
 			}
-			postIndex := iNdEx + intStringLen
+			postIndex := iNdEx + msglen
 			if postIndex < 0 {
 				return ErrInvalidLengthHakeeper
 			}
 			if postIndex > l {
 				return io.ErrUnexpectedEOF
 			}
-			m.UUID = string(dAtA[iNdEx:postIndex])
+			if err := m.ClusterInfo.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
 			iNdEx = postIndex
-		case 2:
+		case 3:
 			if wireType != 2 {
-				return fmt.Errorf("proto: wrong wireType = %d for field ConfigChange", wireType)
+				return fmt.Errorf("proto: wrong wireType = %d for field DNState", wireType)
 			}
 			var msglen int
 			for shift := uint(0); ; shift += 7 {
@@ -937,15 +3282,15 @@ func (m *ScheduleCommand) Unmarshal(dAtA []byte) error {
 			if postIndex > l {
 				return io.ErrUnexpectedEOF
 			}
-			if err := m.ConfigChange.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+			if err := m.DNState.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
 				return err
 			}
 			iNdEx = postIndex
-		case 3:
-			if wireType != 0 {
-				return fmt.Errorf("proto: wrong wireType = %d for field ServiceType", wireType)
+		case 4:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field LogState", wireType)
 			}
-			m.ServiceType = 0
+			var msglen int
 			for shift := uint(0); ; shift += 7 {
 				if shift >= 64 {
 					return ErrIntOverflowHakeeper
@@ -955,11 +3300,25 @@ func (m *ScheduleCommand) Unmarshal(dAtA []byte) error {
 				}
 				b := dAtA[iNdEx]
 				iNdEx++
-				m.ServiceType |= ServiceType(b&0x7F) << shift
+				msglen |= int(b&0x7F) << shift
 				if b < 0x80 {
 					break
 				}
 			}
+			if msglen < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			postIndex := iNdEx + msglen
+			if postIndex < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			if err := m.LogState.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
+			iNdEx = postIndex
 		default:
 			iNdEx = preIndex
 			skippy, err := skipHakeeper(dAtA[iNdEx:])
@@ -982,7 +3341,7 @@ func (m *ScheduleCommand) Unmarshal(dAtA []byte) error {
 	}
 	return nil
 }
-func (m *CommandBatch) Unmarshal(dAtA []byte) error {
+func (m *RSMState) Unmarshal(dAtA []byte) error {
 	l := len(dAtA)
 	iNdEx := 0
 	for iNdEx < l {
@@ -1005,13 +3364,51 @@ func (m *CommandBatch) Unmarshal(dAtA []byte) error {
 		fieldNum := int32(wire >> 3)
 		wireType := int(wire & 0x7)
 		if wireType == 4 {
-			return fmt.Errorf("proto: CommandBatch: wiretype end group for non-group")
+			return fmt.Errorf("proto: RSMState: wiretype end group for non-group")
 		}
 		if fieldNum <= 0 {
-			return fmt.Errorf("proto: CommandBatch: illegal tag %d (wire type %d)", fieldNum, wire)
+			return fmt.Errorf("proto: RSMState: illegal tag %d (wire type %d)", fieldNum, wire)
 		}
 		switch fieldNum {
 		case 1:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Tick", wireType)
+			}
+			m.Tick = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowHakeeper
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.Tick |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+		case 2:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field NextID", wireType)
+			}
+			m.NextID = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowHakeeper
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.NextID |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+		case 3:
 			if wireType != 0 {
 				return fmt.Errorf("proto: wrong wireType = %d for field Term", wireType)
 			}
@@ -1030,9 +3427,9 @@ func (m *CommandBatch) Unmarshal(dAtA []byte) error {
 					break
 				}
 			}
-		case 2:
+		case 4:
 			if wireType != 2 {
-				return fmt.Errorf("proto: wrong wireType = %d for field Commands", wireType)
+				return fmt.Errorf("proto: wrong wireType = %d for field ScheduleCommands", wireType)
 			}
 			var msglen int
 			for shift := uint(0); ; shift += 7 {
@@ -1059,8 +3456,315 @@ func (m *CommandBatch) Unmarshal(dAtA []byte) error {
 			if postIndex > l {
 				return io.ErrUnexpectedEOF
 			}
-			m.Commands = append(m.Commands, ScheduleCommand{})
-			if err := m.Commands[len(m.Commands)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+			if m.ScheduleCommands == nil {
+				m.ScheduleCommands = make(map[string]CommandBatch)
+			}
+			var mapkey string
+			mapvalue := &CommandBatch{}
+			for iNdEx < postIndex {
+				entryPreIndex := iNdEx
+				var wire uint64
+				for shift := uint(0); ; shift += 7 {
+					if shift >= 64 {
+						return ErrIntOverflowHakeeper
+					}
+					if iNdEx >= l {
+						return io.ErrUnexpectedEOF
+					}
+					b := dAtA[iNdEx]
+					iNdEx++
+					wire |= uint64(b&0x7F) << shift
+					if b < 0x80 {
+						break
+					}
+				}
+				fieldNum := int32(wire >> 3)
+				if fieldNum == 1 {
+					var stringLenmapkey uint64
+					for shift := uint(0); ; shift += 7 {
+						if shift >= 64 {
+							return ErrIntOverflowHakeeper
+						}
+						if iNdEx >= l {
+							return io.ErrUnexpectedEOF
+						}
+						b := dAtA[iNdEx]
+						iNdEx++
+						stringLenmapkey |= uint64(b&0x7F) << shift
+						if b < 0x80 {
+							break
+						}
+					}
+					intStringLenmapkey := int(stringLenmapkey)
+					if intStringLenmapkey < 0 {
+						return ErrInvalidLengthHakeeper
+					}
+					postStringIndexmapkey := iNdEx + intStringLenmapkey
+					if postStringIndexmapkey < 0 {
+						return ErrInvalidLengthHakeeper
+					}
+					if postStringIndexmapkey > l {
+						return io.ErrUnexpectedEOF
+					}
+					mapkey = string(dAtA[iNdEx:postStringIndexmapkey])
+					iNdEx = postStringIndexmapkey
+				} else if fieldNum == 2 {
+					var mapmsglen int
+					for shift := uint(0); ; shift += 7 {
+						if shift >= 64 {
+							return ErrIntOverflowHakeeper
+						}
+						if iNdEx >= l {
+							return io.ErrUnexpectedEOF
+						}
+						b := dAtA[iNdEx]
+						iNdEx++
+						mapmsglen |= int(b&0x7F) << shift
+						if b < 0x80 {
+							break
+						}
+					}
+					if mapmsglen < 0 {
+						return ErrInvalidLengthHakeeper
+					}
+					postmsgIndex := iNdEx + mapmsglen
+					if postmsgIndex < 0 {
+						return ErrInvalidLengthHakeeper
+					}
+					if postmsgIndex > l {
+						return io.ErrUnexpectedEOF
+					}
+					mapvalue = &CommandBatch{}
+					if err := mapvalue.Unmarshal(dAtA[iNdEx:postmsgIndex]); err != nil {
+						return err
+					}
+					iNdEx = postmsgIndex
+				} else {
+					iNdEx = entryPreIndex
+					skippy, err := skipHakeeper(dAtA[iNdEx:])
+					if err != nil {
+						return err
+					}
+					if (skippy < 0) || (iNdEx+skippy) < 0 {
+						return ErrInvalidLengthHakeeper
+					}
+					if (iNdEx + skippy) > postIndex {
+						return io.ErrUnexpectedEOF
+					}
+					iNdEx += skippy
+				}
+			}
+			m.ScheduleCommands[mapkey] = *mapvalue
+			iNdEx = postIndex
+		case 5:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field LogShards", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowHakeeper
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			postIndex := iNdEx + msglen
+			if postIndex < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			if m.LogShards == nil {
+				m.LogShards = make(map[string]uint64)
+			}
+			var mapkey string
+			var mapvalue uint64
+			for iNdEx < postIndex {
+				entryPreIndex := iNdEx
+				var wire uint64
+				for shift := uint(0); ; shift += 7 {
+					if shift >= 64 {
+						return ErrIntOverflowHakeeper
+					}
+					if iNdEx >= l {
+						return io.ErrUnexpectedEOF
+					}
+					b := dAtA[iNdEx]
+					iNdEx++
+					wire |= uint64(b&0x7F) << shift
+					if b < 0x80 {
+						break
+					}
+				}
+				fieldNum := int32(wire >> 3)
+				if fieldNum == 1 {
+					var stringLenmapkey uint64
+					for shift := uint(0); ; shift += 7 {
+						if shift >= 64 {
+							return ErrIntOverflowHakeeper
+						}
+						if iNdEx >= l {
+							return io.ErrUnexpectedEOF
+						}
+						b := dAtA[iNdEx]
+						iNdEx++
+						stringLenmapkey |= uint64(b&0x7F) << shift
+						if b < 0x80 {
+							break
+						}
+					}
+					intStringLenmapkey := int(stringLenmapkey)
+					if intStringLenmapkey < 0 {
+						return ErrInvalidLengthHakeeper
+					}
+					postStringIndexmapkey := iNdEx + intStringLenmapkey
+					if postStringIndexmapkey < 0 {
+						return ErrInvalidLengthHakeeper
+					}
+					if postStringIndexmapkey > l {
+						return io.ErrUnexpectedEOF
+					}
+					mapkey = string(dAtA[iNdEx:postStringIndexmapkey])
+					iNdEx = postStringIndexmapkey
+				} else if fieldNum == 2 {
+					for shift := uint(0); ; shift += 7 {
+						if shift >= 64 {
+							return ErrIntOverflowHakeeper
+						}
+						if iNdEx >= l {
+							return io.ErrUnexpectedEOF
+						}
+						b := dAtA[iNdEx]
+						iNdEx++
+						mapvalue |= uint64(b&0x7F) << shift
+						if b < 0x80 {
+							break
+						}
+					}
+				} else {
+					iNdEx = entryPreIndex
+					skippy, err := skipHakeeper(dAtA[iNdEx:])
+					if err != nil {
+						return err
+					}
+					if (skippy < 0) || (iNdEx+skippy) < 0 {
+						return ErrInvalidLengthHakeeper
+					}
+					if (iNdEx + skippy) > postIndex {
+						return io.ErrUnexpectedEOF
+					}
+					iNdEx += skippy
+				}
+			}
+			m.LogShards[mapkey] = mapvalue
+			iNdEx = postIndex
+		case 6:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field DNState", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowHakeeper
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			postIndex := iNdEx + msglen
+			if postIndex < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			if err := m.DNState.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
+			iNdEx = postIndex
+		case 7:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field LogState", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowHakeeper
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			postIndex := iNdEx + msglen
+			if postIndex < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			if err := m.LogState.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
+			iNdEx = postIndex
+		case 8:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field ClusterInfo", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowHakeeper
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			postIndex := iNdEx + msglen
+			if postIndex < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			if err := m.ClusterInfo.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
 				return err
 			}
 			iNdEx = postIndex
diff --git a/pkg/pb/logservice/logservice.pb.go b/pkg/pb/logservice/logservice.pb.go
index d8317e63a..05a120759 100644
--- a/pkg/pb/logservice/logservice.pb.go
+++ b/pkg/pb/logservice/logservice.pb.go
@@ -127,8 +127,9 @@ func (ErrorCode) EnumDescriptor() ([]byte, []int) {
 	return fileDescriptor_fd1040c5381ab5a7, []int{1}
 }
 
-// LogShardInfo contains information of a launched shard.
+// LogShardInfo contains information a log shard.
 type LogShardInfo struct {
+	// ShardID is the ID of a Log shard.
 	ShardID uint64 `protobuf:"varint,1,opt,name=ShardID,proto3" json:"ShardID,omitempty"`
 	// Replicas is a map of ReplicaID to LogStore UUID, it describe the member
 	// replicas of the shard at the given Epoch.
@@ -214,6 +215,56 @@ func (m *LogShardInfo) GetTerm() uint64 {
 	return 0
 }
 
+// LogReplicaInfo contains information of a log replica.
+type LogReplicaInfo struct {
+	LogShardInfo `protobuf:"bytes,1,opt,name=LogShardInfo,proto3,embedded=LogShardInfo" json:"LogShardInfo"`
+	// ReplciaID is the ID of a replica within the Log shard.
+	ReplicaID            uint64   `protobuf:"varint,2,opt,name=ReplicaID,proto3" json:"ReplicaID,omitempty"`
+	XXX_NoUnkeyedLiteral struct{} `json:"-"`
+	XXX_unrecognized     []byte   `json:"-"`
+	XXX_sizecache        int32    `json:"-"`
+}
+
+func (m *LogReplicaInfo) Reset()         { *m = LogReplicaInfo{} }
+func (m *LogReplicaInfo) String() string { return proto.CompactTextString(m) }
+func (*LogReplicaInfo) ProtoMessage()    {}
+func (*LogReplicaInfo) Descriptor() ([]byte, []int) {
+	return fileDescriptor_fd1040c5381ab5a7, []int{1}
+}
+func (m *LogReplicaInfo) XXX_Unmarshal(b []byte) error {
+	return m.Unmarshal(b)
+}
+func (m *LogReplicaInfo) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	if deterministic {
+		return xxx_messageInfo_LogReplicaInfo.Marshal(b, m, deterministic)
+	} else {
+		b = b[:cap(b)]
+		n, err := m.MarshalToSizedBuffer(b)
+		if err != nil {
+			return nil, err
+		}
+		return b[:n], nil
+	}
+}
+func (m *LogReplicaInfo) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_LogReplicaInfo.Merge(m, src)
+}
+func (m *LogReplicaInfo) XXX_Size() int {
+	return m.Size()
+}
+func (m *LogReplicaInfo) XXX_DiscardUnknown() {
+	xxx_messageInfo_LogReplicaInfo.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_LogReplicaInfo proto.InternalMessageInfo
+
+func (m *LogReplicaInfo) GetReplicaID() uint64 {
+	if m != nil {
+		return m.ReplicaID
+	}
+	return 0
+}
+
 // LogStoreHeartbeat is the periodic message sent to the HAKeeper by Log Stores.
 type LogStoreHeartbeat struct {
 	// UUID is the uuid of the Log Store.
@@ -221,21 +272,21 @@ type LogStoreHeartbeat struct {
 	RaftAddress    string `protobuf:"bytes,2,opt,name=RaftAddress,proto3" json:"RaftAddress,omitempty"`
 	ServiceAddress string `protobuf:"bytes,3,opt,name=ServiceAddress,proto3" json:"ServiceAddress,omitempty"`
 	GossipAddress  string `protobuf:"bytes,4,opt,name=GossipAddress,proto3" json:"GossipAddress,omitempty"`
-	// Shards is a list of LogShardInfo instances collected on the specified
+	// Replicas is a list of LogReplicaInfo instances collected on the specified
 	// LogStore. Details in Shards are based on the local knowledge of each
 	// replica running on the current LogStore, it may not be accurate or
 	// update to date due to various reasons.
-	Shards               []LogShardInfo `protobuf:"bytes,5,rep,name=Shards,proto3" json:"Shards"`
-	XXX_NoUnkeyedLiteral struct{}       `json:"-"`
-	XXX_unrecognized     []byte         `json:"-"`
-	XXX_sizecache        int32          `json:"-"`
+	Replicas             []LogReplicaInfo `protobuf:"bytes,5,rep,name=Replicas,proto3" json:"Replicas"`
+	XXX_NoUnkeyedLiteral struct{}         `json:"-"`
+	XXX_unrecognized     []byte           `json:"-"`
+	XXX_sizecache        int32            `json:"-"`
 }
 
 func (m *LogStoreHeartbeat) Reset()         { *m = LogStoreHeartbeat{} }
 func (m *LogStoreHeartbeat) String() string { return proto.CompactTextString(m) }
 func (*LogStoreHeartbeat) ProtoMessage()    {}
 func (*LogStoreHeartbeat) Descriptor() ([]byte, []int) {
-	return fileDescriptor_fd1040c5381ab5a7, []int{1}
+	return fileDescriptor_fd1040c5381ab5a7, []int{2}
 }
 func (m *LogStoreHeartbeat) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -292,9 +343,9 @@ func (m *LogStoreHeartbeat) GetGossipAddress() string {
 	return ""
 }
 
-func (m *LogStoreHeartbeat) GetShards() []LogShardInfo {
+func (m *LogStoreHeartbeat) GetReplicas() []LogReplicaInfo {
 	if m != nil {
-		return m.Shards
+		return m.Replicas
 	}
 	return nil
 }
@@ -317,7 +368,7 @@ func (m *DNShardInfo) Reset()         { *m = DNShardInfo{} }
 func (m *DNShardInfo) String() string { return proto.CompactTextString(m) }
 func (*DNShardInfo) ProtoMessage()    {}
 func (*DNShardInfo) Descriptor() ([]byte, []int) {
-	return fileDescriptor_fd1040c5381ab5a7, []int{2}
+	return fileDescriptor_fd1040c5381ab5a7, []int{3}
 }
 func (m *DNShardInfo) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -376,7 +427,7 @@ func (m *DNStoreHeartbeat) Reset()         { *m = DNStoreHeartbeat{} }
 func (m *DNStoreHeartbeat) String() string { return proto.CompactTextString(m) }
 func (*DNStoreHeartbeat) ProtoMessage()    {}
 func (*DNStoreHeartbeat) Descriptor() ([]byte, []int) {
-	return fileDescriptor_fd1040c5381ab5a7, []int{3}
+	return fileDescriptor_fd1040c5381ab5a7, []int{4}
 }
 func (m *DNStoreHeartbeat) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -419,6 +470,77 @@ func (m *DNStoreHeartbeat) GetShards() []DNShardInfo {
 	return nil
 }
 
+type RSMState struct {
+	Index                uint64            `protobuf:"varint,1,opt,name=Index,proto3" json:"Index,omitempty"`
+	LeaseHolderID        uint64            `protobuf:"varint,2,opt,name=LeaseHolderID,proto3" json:"LeaseHolderID,omitempty"`
+	TruncatedIndex       uint64            `protobuf:"varint,3,opt,name=TruncatedIndex,proto3" json:"TruncatedIndex,omitempty"`
+	LeaseHistory         map[uint64]uint64 `protobuf:"bytes,4,rep,name=LeaseHistory,proto3" json:"LeaseHistory,omitempty" protobuf_key:"varint,1,opt,name=key,proto3" protobuf_val:"varint,2,opt,name=value,proto3"`
+	XXX_NoUnkeyedLiteral struct{}          `json:"-"`
+	XXX_unrecognized     []byte            `json:"-"`
+	XXX_sizecache        int32             `json:"-"`
+}
+
+func (m *RSMState) Reset()         { *m = RSMState{} }
+func (m *RSMState) String() string { return proto.CompactTextString(m) }
+func (*RSMState) ProtoMessage()    {}
+func (*RSMState) Descriptor() ([]byte, []int) {
+	return fileDescriptor_fd1040c5381ab5a7, []int{5}
+}
+func (m *RSMState) XXX_Unmarshal(b []byte) error {
+	return m.Unmarshal(b)
+}
+func (m *RSMState) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	if deterministic {
+		return xxx_messageInfo_RSMState.Marshal(b, m, deterministic)
+	} else {
+		b = b[:cap(b)]
+		n, err := m.MarshalToSizedBuffer(b)
+		if err != nil {
+			return nil, err
+		}
+		return b[:n], nil
+	}
+}
+func (m *RSMState) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_RSMState.Merge(m, src)
+}
+func (m *RSMState) XXX_Size() int {
+	return m.Size()
+}
+func (m *RSMState) XXX_DiscardUnknown() {
+	xxx_messageInfo_RSMState.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_RSMState proto.InternalMessageInfo
+
+func (m *RSMState) GetIndex() uint64 {
+	if m != nil {
+		return m.Index
+	}
+	return 0
+}
+
+func (m *RSMState) GetLeaseHolderID() uint64 {
+	if m != nil {
+		return m.LeaseHolderID
+	}
+	return 0
+}
+
+func (m *RSMState) GetTruncatedIndex() uint64 {
+	if m != nil {
+		return m.TruncatedIndex
+	}
+	return 0
+}
+
+func (m *RSMState) GetLeaseHistory() map[uint64]uint64 {
+	if m != nil {
+		return m.LeaseHistory
+	}
+	return nil
+}
+
 type LogRecord struct {
 	Index                uint64   `protobuf:"varint,1,opt,name=Index,proto3" json:"Index,omitempty"`
 	Data                 []byte   `protobuf:"bytes,2,opt,name=Data,proto3" json:"Data,omitempty"`
@@ -431,7 +553,7 @@ func (m *LogRecord) Reset()         { *m = LogRecord{} }
 func (m *LogRecord) String() string { return proto.CompactTextString(m) }
 func (*LogRecord) ProtoMessage()    {}
 func (*LogRecord) Descriptor() ([]byte, []int) {
-	return fileDescriptor_fd1040c5381ab5a7, []int{4}
+	return fileDescriptor_fd1040c5381ab5a7, []int{6}
 }
 func (m *LogRecord) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -493,7 +615,7 @@ func (m *Request) Reset()         { *m = Request{} }
 func (m *Request) String() string { return proto.CompactTextString(m) }
 func (*Request) ProtoMessage()    {}
 func (*Request) Descriptor() ([]byte, []int) {
-	return fileDescriptor_fd1040c5381ab5a7, []int{5}
+	return fileDescriptor_fd1040c5381ab5a7, []int{7}
 }
 func (m *Request) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -602,7 +724,7 @@ func (m *Response) Reset()         { *m = Response{} }
 func (m *Response) String() string { return proto.CompactTextString(m) }
 func (*Response) ProtoMessage()    {}
 func (*Response) Descriptor() ([]byte, []int) {
-	return fileDescriptor_fd1040c5381ab5a7, []int{6}
+	return fileDescriptor_fd1040c5381ab5a7, []int{8}
 }
 func (m *Response) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -691,7 +813,7 @@ func (m *LogRecordResponse) Reset()         { *m = LogRecordResponse{} }
 func (m *LogRecordResponse) String() string { return proto.CompactTextString(m) }
 func (*LogRecordResponse) ProtoMessage()    {}
 func (*LogRecordResponse) Descriptor() ([]byte, []int) {
-	return fileDescriptor_fd1040c5381ab5a7, []int{7}
+	return fileDescriptor_fd1040c5381ab5a7, []int{9}
 }
 func (m *LogRecordResponse) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -732,9 +854,12 @@ func init() {
 	proto.RegisterEnum("logservice.ErrorCode", ErrorCode_name, ErrorCode_value)
 	proto.RegisterType((*LogShardInfo)(nil), "logservice.LogShardInfo")
 	proto.RegisterMapType((map[uint64]string)(nil), "logservice.LogShardInfo.ReplicasEntry")
+	proto.RegisterType((*LogReplicaInfo)(nil), "logservice.LogReplicaInfo")
 	proto.RegisterType((*LogStoreHeartbeat)(nil), "logservice.LogStoreHeartbeat")
 	proto.RegisterType((*DNShardInfo)(nil), "logservice.DNShardInfo")
 	proto.RegisterType((*DNStoreHeartbeat)(nil), "logservice.DNStoreHeartbeat")
+	proto.RegisterType((*RSMState)(nil), "logservice.RSMState")
+	proto.RegisterMapType((map[uint64]uint64)(nil), "logservice.RSMState.LeaseHistoryEntry")
 	proto.RegisterType((*LogRecord)(nil), "logservice.LogRecord")
 	proto.RegisterType((*Request)(nil), "logservice.Request")
 	proto.RegisterType((*Response)(nil), "logservice.Response")
@@ -744,61 +869,69 @@ func init() {
 func init() { proto.RegisterFile("logservice.proto", fileDescriptor_fd1040c5381ab5a7) }
 
 var fileDescriptor_fd1040c5381ab5a7 = []byte{
-	// 858 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x94, 0x55, 0x4f, 0x6b, 0xe3, 0xc6,
-	0x1b, 0x5e, 0xd9, 0xb2, 0x65, 0xbf, 0x71, 0xcc, 0xec, 0xf0, 0xcb, 0xfe, 0xd4, 0xb0, 0xa4, 0xc6,
-	0x94, 0x25, 0x2c, 0xd4, 0x0b, 0x59, 0x52, 0x4a, 0x7b, 0x72, 0x2c, 0xb1, 0xeb, 0xe2, 0xc8, 0x61,
-	0xac, 0x1c, 0x7a, 0x28, 0xcb, 0xc4, 0x7a, 0xfd, 0xa7, 0x95, 0x3d, 0xae, 0x34, 0x0e, 0x71, 0xaf,
-	0xa5, 0xd7, 0x7e, 0xae, 0xbd, 0xb5, 0x9f, 0xa0, 0x94, 0x9c, 0x7a, 0xeb, 0x57, 0x28, 0x33, 0x23,
-	0x29, 0x72, 0xca, 0xc2, 0xf6, 0xf6, 0x3e, 0xcf, 0xfb, 0xff, 0x99, 0x57, 0x08, 0x48, 0x2c, 0xe6,
-	0x29, 0x26, 0xb7, 0xcb, 0x29, 0xf6, 0x36, 0x89, 0x90, 0x82, 0xc2, 0x03, 0x73, 0xfc, 0xf9, 0x7c,
-	0x29, 0x17, 0xdb, 0x9b, 0xde, 0x54, 0xac, 0x5e, 0xcd, 0xc5, 0x5c, 0xbc, 0xd2, 0x21, 0x37, 0xdb,
-	0x99, 0x46, 0x1a, 0x68, 0xcb, 0xa4, 0x76, 0xff, 0xb6, 0xa0, 0x35, 0x12, 0xf3, 0xc9, 0x82, 0x27,
-	0xd1, 0x70, 0x3d, 0x13, 0xd4, 0x05, 0xc7, 0x00, 0xcf, 0xb5, 0x3a, 0xd6, 0xa9, 0xcd, 0x72, 0x48,
-	0x2f, 0xa0, 0xc1, 0x70, 0x13, 0x2f, 0xa7, 0x3c, 0x75, 0x2b, 0x9d, 0xea, 0xe9, 0xc1, 0xd9, 0x8b,
-	0x5e, 0x69, 0x94, 0x72, 0x95, 0x5e, 0x1e, 0xe8, 0xaf, 0x65, 0xb2, 0x63, 0x45, 0x1e, 0xfd, 0x1f,
-	0xd4, 0xfc, 0x8d, 0x98, 0x2e, 0xdc, 0xaa, 0xae, 0x6d, 0x00, 0x3d, 0x86, 0xc6, 0x08, 0x79, 0x84,
-	0xc9, 0xd0, 0x73, 0x6d, 0xed, 0x28, 0x30, 0xa5, 0x60, 0x87, 0x98, 0xac, 0xdc, 0x9a, 0xe6, 0xb5,
-	0x7d, 0xfc, 0x35, 0x1c, 0xee, 0x35, 0xa0, 0x04, 0xaa, 0x3f, 0xe0, 0x2e, 0x1b, 0x58, 0x99, 0xaa,
-	0xd1, 0x2d, 0x8f, 0xb7, 0xe8, 0x56, 0x3a, 0xd6, 0x69, 0x93, 0x19, 0xf0, 0x55, 0xe5, 0x4b, 0xab,
-	0xfb, 0x9b, 0x05, 0x4f, 0xd5, 0xac, 0x52, 0x24, 0xf8, 0x16, 0x79, 0x22, 0x6f, 0x90, 0x4b, 0xd5,
-	0xe6, 0xfa, 0x3a, 0xdb, 0xb9, 0xc9, 0xb4, 0x4d, 0x3b, 0x70, 0xc0, 0xf8, 0x4c, 0xf6, 0xa3, 0x28,
-	0xc1, 0x34, 0xcd, 0x2a, 0x95, 0x29, 0xfa, 0x02, 0xda, 0x13, 0xb3, 0x7e, 0x1e, 0x54, 0xd5, 0x41,
-	0x8f, 0x58, 0xfa, 0x19, 0x1c, 0xbe, 0x11, 0x69, 0xba, 0xdc, 0xe4, 0x61, 0xb6, 0x0e, 0xdb, 0x27,
-	0xe9, 0x17, 0x50, 0xd7, 0x0a, 0xa6, 0x6e, 0x4d, 0xcb, 0xeb, 0x7e, 0x48, 0xde, 0x0b, 0xfb, 0xfd,
-	0x1f, 0x9f, 0x3e, 0x61, 0x59, 0x74, 0xd7, 0x87, 0x03, 0x2f, 0xf8, 0x98, 0x17, 0x7c, 0x0e, 0xcd,
-	0x4c, 0xb7, 0xa1, 0xa7, 0xd7, 0xb1, 0xd9, 0x03, 0xd1, 0xfd, 0x0e, 0x88, 0x17, 0x7c, 0x84, 0x2c,
-	0xe7, 0xc5, 0x98, 0xe6, 0x0a, 0xfe, 0x5f, 0x1e, 0xb3, 0x34, 0xc8, 0xa3, 0x29, 0xcf, 0xa1, 0x39,
-	0x12, 0x73, 0x86, 0x53, 0x91, 0x44, 0xea, 0x79, 0x86, 0xeb, 0x08, 0xef, 0xb2, 0x09, 0x0d, 0x50,
-	0xdd, 0x3c, 0x2e, 0xb9, 0x1e, 0xad, 0xc5, 0xb4, 0xdd, 0xfd, 0xa5, 0x02, 0x0e, 0xc3, 0x1f, 0xb7,
-	0x98, 0x4a, 0xda, 0x83, 0xfa, 0x25, 0xca, 0x85, 0x88, 0x74, 0x5a, 0xfb, 0xec, 0x59, 0xb9, 0xb3,
-	0xf1, 0x84, 0xbb, 0x0d, 0xb2, 0x2c, 0x4a, 0xd5, 0x0b, 0xf8, 0x2a, 0xbf, 0x01, 0x6d, 0x97, 0xd5,
-	0xa9, 0xee, 0xab, 0x53, 0xcc, 0x64, 0x97, 0x67, 0x72, 0xc1, 0xb9, 0xe4, 0x77, 0x93, 0xe5, 0x4f,
-	0x98, 0x9d, 0x60, 0x0e, 0x95, 0x27, 0x5c, 0xae, 0x50, 0x6c, 0xa5, 0x5b, 0xef, 0x58, 0xa7, 0x55,
-	0x96, 0x43, 0xa5, 0x73, 0xae, 0x83, 0xe7, 0x3a, 0x46, 0xe7, 0x82, 0xd0, 0x5b, 0x06, 0x43, 0xcf,
-	0x6d, 0x98, 0x8b, 0x56, 0xb6, 0x3a, 0xb5, 0x2b, 0xbe, 0x8b, 0x05, 0x8f, 0x74, 0xa7, 0xa6, 0x76,
-	0x95, 0xa9, 0xee, 0xcf, 0x15, 0xf5, 0xf9, 0xa5, 0x1b, 0xb1, 0x4e, 0xf1, 0x3f, 0x0b, 0xf1, 0x1a,
-	0x9a, 0x7e, 0x92, 0x88, 0x64, 0x20, 0x22, 0xa3, 0x46, 0xfb, 0xec, 0xa8, 0x9c, 0x52, 0x38, 0xd9,
-	0x43, 0x1c, 0xed, 0x42, 0x4b, 0x83, 0x4b, 0x4c, 0x53, 0x3e, 0xc7, 0xec, 0xb4, 0xf7, 0xb8, 0xb2,
-	0x9a, 0xf6, 0x07, 0xd4, 0xac, 0x95, 0xd5, 0x7c, 0x0e, 0xcd, 0x11, 0x4f, 0xa5, 0xf1, 0xd4, 0x8d,
-	0x32, 0x05, 0xf1, 0x58, 0x05, 0xe7, 0xdf, 0x2a, 0x7c, 0xa3, 0xbf, 0x5d, 0x73, 0x44, 0x85, 0x1a,
-	0xe7, 0xea, 0x42, 0x14, 0x93, 0xba, 0x96, 0xbe, 0xc8, 0xa3, 0x47, 0x1f, 0x8e, 0xf1, 0x66, 0xf7,
-	0x98, 0xc7, 0xbe, 0xbc, 0x03, 0x78, 0x90, 0x8a, 0x02, 0xd4, 0x07, 0xcc, 0xef, 0x87, 0x3e, 0x79,
-	0x42, 0x0f, 0xc0, 0xf1, 0xfc, 0x49, 0xc8, 0xc6, 0xdf, 0x12, 0x4b, 0x39, 0xfa, 0x57, 0x57, 0x7e,
-	0xe0, 0x91, 0x0a, 0x6d, 0x80, 0xcd, 0xfc, 0xbe, 0x47, 0xaa, 0xb4, 0x05, 0x8d, 0x90, 0x5d, 0x07,
-	0x03, 0x95, 0x60, 0x53, 0x02, 0xad, 0x37, 0x7e, 0xf8, 0xae, 0x60, 0x6a, 0xaa, 0xc4, 0x60, 0x1c,
-	0x04, 0xfe, 0x20, 0x24, 0x75, 0xda, 0x06, 0xc8, 0xc0, 0x3b, 0x36, 0x26, 0xce, 0xcb, 0x5f, 0x2b,
-	0xa5, 0xf7, 0x50, 0xa1, 0x81, 0xd0, 0xd0, 0xb4, 0xce, 0xae, 0x88, 0x58, 0xaa, 0xc9, 0x80, 0xaf,
-	0xa7, 0x18, 0x63, 0x44, 0x2a, 0xaa, 0xc9, 0x70, 0x7d, 0xcb, 0xe3, 0x65, 0xa4, 0x35, 0x26, 0x55,
-	0x4a, 0xa1, 0x9d, 0x31, 0x79, 0x8e, 0x5d, 0xe2, 0x32, 0xdd, 0x48, 0x8d, 0x3e, 0x03, 0xba, 0xcf,
-	0x29, 0x2d, 0x49, 0x5d, 0xd5, 0x67, 0xf8, 0x3d, 0x4e, 0x25, 0x46, 0xc4, 0xa1, 0x4f, 0xe1, 0x50,
-	0x17, 0x0e, 0x84, 0x64, 0xc8, 0xa3, 0x1d, 0x69, 0xa8, 0x96, 0x93, 0x5d, 0x2a, 0x71, 0x35, 0x88,
-	0x45, 0x8a, 0x11, 0x69, 0xd2, 0x4f, 0xe0, 0x48, 0xbf, 0x55, 0x3f, 0x4e, 0x54, 0x4c, 0x98, 0x6c,
-	0xd7, 0x53, 0xae, 0xf2, 0x23, 0xb5, 0xe5, 0x78, 0x2b, 0xc7, 0x33, 0xc6, 0xd7, 0x73, 0x24, 0xa8,
-	0x26, 0x09, 0x84, 0x1c, 0x21, 0x4f, 0xf1, 0xad, 0x88, 0x23, 0x4c, 0xc8, 0x8c, 0x1e, 0x01, 0x19,
-	0xcb, 0x05, 0x26, 0xa6, 0xaa, 0x59, 0xfa, 0x2f, 0xe7, 0xa2, 0xf5, 0xfe, 0xfe, 0xc4, 0xfa, 0xfd,
-	0xfe, 0xc4, 0xfa, 0xf3, 0xfe, 0xc4, 0xba, 0xa9, 0xeb, 0x5f, 0xd3, 0xeb, 0x7f, 0x02, 0x00, 0x00,
-	0xff, 0xff, 0x37, 0x40, 0x45, 0x10, 0xe9, 0x06, 0x00, 0x00,
+	// 990 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x94, 0x56, 0xcf, 0x6b, 0x1b, 0x47,
+	0x14, 0xce, 0x4a, 0xab, 0x5f, 0xcf, 0xb2, 0x18, 0x0f, 0x75, 0xba, 0x35, 0xc1, 0x31, 0xa2, 0x04,
+	0x13, 0xa8, 0x0c, 0x36, 0x86, 0xd2, 0x16, 0x8a, 0xac, 0xdd, 0x26, 0x0e, 0xf2, 0xca, 0x8c, 0xe4,
+	0x43, 0x0b, 0x25, 0x8c, 0xb4, 0xa3, 0xd5, 0x36, 0xd2, 0x8e, 0xba, 0x3b, 0x32, 0x56, 0xaf, 0xa5,
+	0xd7, 0xfe, 0x5d, 0x39, 0xe6, 0xd2, 0x6b, 0x28, 0x3e, 0xf5, 0xd6, 0x3f, 0xa0, 0x97, 0x32, 0x33,
+	0xbb, 0xd2, 0xc8, 0xc6, 0x6d, 0x72, 0x9b, 0xf7, 0xcd, 0xfb, 0xf9, 0xcd, 0xf7, 0x96, 0x05, 0x34,
+	0xe5, 0x61, 0xca, 0x92, 0xeb, 0x68, 0xc4, 0x5a, 0xf3, 0x84, 0x0b, 0x8e, 0x61, 0x8d, 0xec, 0x7d,
+	0x11, 0x46, 0x62, 0xb2, 0x18, 0xb6, 0x46, 0x7c, 0x76, 0x14, 0xf2, 0x90, 0x1f, 0x29, 0x97, 0xe1,
+	0x62, 0xac, 0x2c, 0x65, 0xa8, 0x93, 0x0e, 0x6d, 0xfe, 0x6d, 0x41, 0xbd, 0xcb, 0xc3, 0xfe, 0x84,
+	0x26, 0xc1, 0x79, 0x3c, 0xe6, 0xd8, 0x81, 0x8a, 0x36, 0x5c, 0xc7, 0x3a, 0xb0, 0x0e, 0x6d, 0x92,
+	0x9b, 0xf8, 0x0c, 0xaa, 0x84, 0xcd, 0xa7, 0xd1, 0x88, 0xa6, 0x4e, 0xe1, 0xa0, 0x78, 0xb8, 0x75,
+	0xfc, 0xac, 0x65, 0xb4, 0x62, 0x66, 0x69, 0xe5, 0x8e, 0x5e, 0x2c, 0x92, 0x25, 0x59, 0xc5, 0xe1,
+	0x4f, 0xa0, 0xe4, 0xcd, 0xf9, 0x68, 0xe2, 0x14, 0x55, 0x6e, 0x6d, 0xe0, 0x3d, 0xa8, 0x76, 0x19,
+	0x0d, 0x58, 0x72, 0xee, 0x3a, 0xb6, 0xba, 0x58, 0xd9, 0x18, 0x83, 0x3d, 0x60, 0xc9, 0xcc, 0x29,
+	0x29, 0x5c, 0x9d, 0xf7, 0xbe, 0x86, 0xed, 0x8d, 0x02, 0x18, 0x41, 0xf1, 0x0d, 0x5b, 0x66, 0x0d,
+	0xcb, 0xa3, 0x2c, 0x74, 0x4d, 0xa7, 0x0b, 0xe6, 0x14, 0x0e, 0xac, 0xc3, 0x1a, 0xd1, 0xc6, 0x57,
+	0x85, 0x2f, 0xad, 0xe6, 0x35, 0x34, 0xba, 0x3c, 0xcc, 0xe2, 0xd5, 0xc8, 0xdf, 0x6d, 0x52, 0xa0,
+	0xd2, 0x6c, 0x1d, 0x3b, 0x0f, 0x0d, 0x77, 0x56, 0x7d, 0xfb, 0xfe, 0xe9, 0xa3, 0x77, 0xef, 0x9f,
+	0x5a, 0x64, 0x93, 0xba, 0x27, 0x50, 0xcb, 0xd3, 0xba, 0xaa, 0xae, 0x4d, 0xd6, 0x40, 0xf3, 0x0f,
+	0x0b, 0x76, 0xa4, 0xbb, 0xe0, 0x09, 0x7b, 0xc9, 0x68, 0x22, 0x86, 0x8c, 0x0a, 0x39, 0xde, 0xd5,
+	0x55, 0xc6, 0x75, 0x8d, 0xa8, 0x33, 0x3e, 0x80, 0x2d, 0x42, 0xc7, 0xa2, 0x1d, 0x04, 0x09, 0x4b,
+	0xd3, 0x6c, 0x02, 0x13, 0xc2, 0xcf, 0xa0, 0xd1, 0xd7, 0x9d, 0xe5, 0x4e, 0x45, 0xe5, 0x74, 0x07,
+	0xc5, 0x9f, 0xc3, 0xf6, 0x0b, 0x9e, 0xa6, 0xd1, 0x3c, 0x77, 0xb3, 0x95, 0xdb, 0x26, 0x88, 0xbf,
+	0x31, 0x1e, 0xb6, 0xa4, 0x1e, 0x76, 0xef, 0xce, 0xec, 0x06, 0x5b, 0x67, 0xb6, 0x9c, 0x7e, 0xfd,
+	0xa4, 0x4d, 0x0f, 0xb6, 0x5c, 0xff, 0x43, 0xf4, 0xf3, 0xdf, 0xf4, 0xfc, 0x08, 0xc8, 0xf5, 0x3f,
+	0x80, 0x9c, 0x53, 0x28, 0xab, 0x84, 0xb9, 0x06, 0x3f, 0x35, 0x5b, 0x35, 0x1a, 0xc9, 0xfa, 0xcc,
+	0x9c, 0x9b, 0xff, 0x58, 0x50, 0x25, 0xfd, 0x8b, 0xbe, 0xa0, 0x82, 0x49, 0x71, 0x9c, 0xc7, 0x01,
+	0xbb, 0xc9, 0x3a, 0xd4, 0x86, 0x24, 0xab, 0xcb, 0x68, 0xca, 0x5e, 0xf2, 0xa9, 0x96, 0xa2, 0xee,
+	0x71, 0x13, 0x94, 0xd4, 0x0f, 0x92, 0x45, 0x3c, 0xa2, 0x82, 0x05, 0x3a, 0x89, 0x96, 0xf2, 0x1d,
+	0x14, 0xbf, 0x82, 0xba, 0x0e, 0x8c, 0x52, 0xc1, 0x93, 0xa5, 0x63, 0xdf, 0xdf, 0x98, 0xbc, 0x9f,
+	0x96, 0xe9, 0xa8, 0x37, 0x66, 0x23, 0x76, 0xef, 0x5b, 0xd8, 0xb9, 0xe7, 0xf2, 0x7f, 0x9a, 0xb7,
+	0x4d, 0xcd, 0x9f, 0x42, 0x4d, 0xbd, 0xe2, 0x88, 0x27, 0xc1, 0x03, 0xd3, 0x63, 0xb0, 0x5d, 0x2a,
+	0xa8, 0x8a, 0xad, 0x13, 0x75, 0x6e, 0xfe, 0x56, 0x80, 0x0a, 0x61, 0x3f, 0x2f, 0x58, 0x2a, 0x70,
+	0x0b, 0xca, 0x17, 0x4c, 0x4c, 0x78, 0xa0, 0xc2, 0x1a, 0xc7, 0x8f, 0xcd, 0x49, 0xf4, 0xcd, 0x60,
+	0x39, 0x67, 0x24, 0xf3, 0x92, 0xf9, 0x7c, 0x3a, 0xcb, 0xf7, 0x4f, 0x9d, 0x4d, 0x6d, 0x14, 0x37,
+	0xb5, 0xb1, 0xea, 0xc9, 0x36, 0x7b, 0x72, 0xa0, 0x72, 0x41, 0x6f, 0xfa, 0xd1, 0x2f, 0x2c, 0x5b,
+	0xff, 0xdc, 0x94, 0x37, 0x83, 0x68, 0xc6, 0xf8, 0x42, 0x38, 0xe5, 0x03, 0xeb, 0xb0, 0x48, 0x72,
+	0x53, 0xaa, 0x2c, 0x57, 0x81, 0xeb, 0x54, 0xb4, 0xca, 0x56, 0x80, 0x9a, 0xd2, 0x3f, 0x77, 0x9d,
+	0xaa, 0xfe, 0x9a, 0xc8, 0xb3, 0x5c, 0xb7, 0x4b, 0xba, 0x9c, 0x72, 0x1a, 0xa8, 0x4a, 0x35, 0x75,
+	0x65, 0x42, 0xcd, 0x5f, 0x0b, 0x72, 0x43, 0xd2, 0x39, 0x8f, 0x53, 0xf6, 0xd1, 0x44, 0x9c, 0x40,
+	0xcd, 0x4b, 0x12, 0x9e, 0x74, 0x78, 0xa0, 0xd9, 0x68, 0x1c, 0xef, 0x9a, 0x21, 0xab, 0x4b, 0xb2,
+	0xf6, 0xc3, 0x4d, 0xa8, 0x2b, 0xe3, 0x82, 0xa5, 0x29, 0x0d, 0x59, 0xb6, 0xde, 0x1b, 0x98, 0xc9,
+	0xa6, 0xfd, 0x00, 0x9b, 0x25, 0x93, 0xcd, 0x27, 0x50, 0xeb, 0xd2, 0x54, 0xe8, 0x9b, 0xb2, 0x66,
+	0x66, 0x05, 0xdc, 0x65, 0xa1, 0x72, 0x9f, 0x85, 0x57, 0xea, 0xfb, 0xa5, 0x45, 0xb4, 0x62, 0xe3,
+	0x54, 0x2a, 0x44, 0x22, 0xa9, 0x63, 0x29, 0x85, 0xef, 0xde, 0xfb, 0x74, 0xc8, 0xdb, 0x6c, 0x1b,
+	0x73, 0xdf, 0xe7, 0x37, 0x00, 0x6b, 0xaa, 0x30, 0x40, 0xb9, 0x43, 0xbc, 0xf6, 0xc0, 0x43, 0x8f,
+	0xf0, 0x16, 0x54, 0x5c, 0xaf, 0x3f, 0x20, 0xbd, 0xef, 0x91, 0x25, 0x2f, 0xda, 0x97, 0x97, 0x9e,
+	0xef, 0xa2, 0x02, 0xae, 0x82, 0x4d, 0xbc, 0xb6, 0x8b, 0x8a, 0xb8, 0x0e, 0xd5, 0x01, 0xb9, 0xf2,
+	0x3b, 0x32, 0xc0, 0xc6, 0x08, 0xea, 0x2f, 0xbc, 0xc1, 0xeb, 0x15, 0x52, 0x92, 0x29, 0x3a, 0x3d,
+	0xdf, 0xf7, 0x3a, 0x03, 0x54, 0xc6, 0x0d, 0x80, 0xcc, 0x78, 0x4d, 0x7a, 0xa8, 0xf2, 0xfc, 0xf7,
+	0x82, 0xf1, 0x1e, 0xd2, 0xd5, 0xe7, 0xca, 0xd4, 0xa5, 0x33, 0x15, 0x21, 0x4b, 0x16, 0xe9, 0xd0,
+	0x78, 0xc4, 0xa6, 0x2c, 0x40, 0x05, 0x59, 0xe4, 0x3c, 0xbe, 0xa6, 0xd3, 0x28, 0x50, 0x1c, 0xa3,
+	0x22, 0xc6, 0xd0, 0xc8, 0x90, 0x3c, 0xc6, 0x36, 0xb0, 0x8c, 0x37, 0x54, 0xc2, 0x8f, 0x01, 0x6f,
+	0x62, 0x92, 0x4b, 0x54, 0x96, 0xf9, 0x09, 0xfb, 0x89, 0x8d, 0x04, 0x0b, 0x50, 0x05, 0xef, 0xc0,
+	0xb6, 0x4a, 0xec, 0x73, 0x41, 0x18, 0x0d, 0x96, 0xa8, 0x2a, 0x4b, 0xf6, 0x97, 0xa9, 0x60, 0xb3,
+	0xce, 0x94, 0xa7, 0x2c, 0x40, 0x35, 0xfc, 0x19, 0xec, 0xaa, 0xb7, 0x6a, 0x4f, 0x13, 0xe9, 0xb3,
+	0xfa, 0xe0, 0xa0, 0x40, 0x4e, 0xd9, 0x5b, 0x88, 0xde, 0x98, 0xd0, 0x38, 0x64, 0x88, 0xc9, 0x4e,
+	0x7c, 0x2e, 0x8c, 0x2f, 0x17, 0x1a, 0xe3, 0x5d, 0x40, 0x3d, 0x31, 0x61, 0x89, 0xce, 0xaa, 0x87,
+	0xfe, 0xab, 0x72, 0xd6, 0x7e, 0x7b, 0xbb, 0x6f, 0xbd, 0xbb, 0xdd, 0xb7, 0xfe, 0xbc, 0xdd, 0xb7,
+	0x7e, 0x38, 0x31, 0x7e, 0x1f, 0x66, 0x54, 0x24, 0xd1, 0x0d, 0x4f, 0xa2, 0x30, 0x8a, 0x73, 0x23,
+	0x66, 0x47, 0xf3, 0x37, 0xe1, 0xd1, 0x7c, 0x78, 0xb4, 0x7e, 0xe6, 0x61, 0x59, 0xfd, 0x4b, 0x9c,
+	0xfc, 0x1b, 0x00, 0x00, 0xff, 0xff, 0x05, 0x15, 0x41, 0x99, 0x9a, 0x08, 0x00, 0x00,
 }
 
 func (m *LogShardInfo) Marshal() (dAtA []byte, err error) {
@@ -865,6 +998,48 @@ func (m *LogShardInfo) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 	return len(dAtA) - i, nil
 }
 
+func (m *LogReplicaInfo) Marshal() (dAtA []byte, err error) {
+	size := m.Size()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	if err != nil {
+		return nil, err
+	}
+	return dAtA[:n], nil
+}
+
+func (m *LogReplicaInfo) MarshalTo(dAtA []byte) (int, error) {
+	size := m.Size()
+	return m.MarshalToSizedBuffer(dAtA[:size])
+}
+
+func (m *LogReplicaInfo) MarshalToSizedBuffer(dAtA []byte) (int, error) {
+	i := len(dAtA)
+	_ = i
+	var l int
+	_ = l
+	if m.XXX_unrecognized != nil {
+		i -= len(m.XXX_unrecognized)
+		copy(dAtA[i:], m.XXX_unrecognized)
+	}
+	if m.ReplicaID != 0 {
+		i = encodeVarintLogservice(dAtA, i, uint64(m.ReplicaID))
+		i--
+		dAtA[i] = 0x10
+	}
+	{
+		size, err := m.LogShardInfo.MarshalToSizedBuffer(dAtA[:i])
+		if err != nil {
+			return 0, err
+		}
+		i -= size
+		i = encodeVarintLogservice(dAtA, i, uint64(size))
+	}
+	i--
+	dAtA[i] = 0xa
+	return len(dAtA) - i, nil
+}
+
 func (m *LogStoreHeartbeat) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
@@ -889,10 +1064,10 @@ func (m *LogStoreHeartbeat) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 		i -= len(m.XXX_unrecognized)
 		copy(dAtA[i:], m.XXX_unrecognized)
 	}
-	if len(m.Shards) > 0 {
-		for iNdEx := len(m.Shards) - 1; iNdEx >= 0; iNdEx-- {
+	if len(m.Replicas) > 0 {
+		for iNdEx := len(m.Replicas) - 1; iNdEx >= 0; iNdEx-- {
 			{
-				size, err := m.Shards[iNdEx].MarshalToSizedBuffer(dAtA[:i])
+				size, err := m.Replicas[iNdEx].MarshalToSizedBuffer(dAtA[:i])
 				if err != nil {
 					return 0, err
 				}
@@ -1019,6 +1194,63 @@ func (m *DNStoreHeartbeat) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 	return len(dAtA) - i, nil
 }
 
+func (m *RSMState) Marshal() (dAtA []byte, err error) {
+	size := m.Size()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	if err != nil {
+		return nil, err
+	}
+	return dAtA[:n], nil
+}
+
+func (m *RSMState) MarshalTo(dAtA []byte) (int, error) {
+	size := m.Size()
+	return m.MarshalToSizedBuffer(dAtA[:size])
+}
+
+func (m *RSMState) MarshalToSizedBuffer(dAtA []byte) (int, error) {
+	i := len(dAtA)
+	_ = i
+	var l int
+	_ = l
+	if m.XXX_unrecognized != nil {
+		i -= len(m.XXX_unrecognized)
+		copy(dAtA[i:], m.XXX_unrecognized)
+	}
+	if len(m.LeaseHistory) > 0 {
+		for k := range m.LeaseHistory {
+			v := m.LeaseHistory[k]
+			baseI := i
+			i = encodeVarintLogservice(dAtA, i, uint64(v))
+			i--
+			dAtA[i] = 0x10
+			i = encodeVarintLogservice(dAtA, i, uint64(k))
+			i--
+			dAtA[i] = 0x8
+			i = encodeVarintLogservice(dAtA, i, uint64(baseI-i))
+			i--
+			dAtA[i] = 0x22
+		}
+	}
+	if m.TruncatedIndex != 0 {
+		i = encodeVarintLogservice(dAtA, i, uint64(m.TruncatedIndex))
+		i--
+		dAtA[i] = 0x18
+	}
+	if m.LeaseHolderID != 0 {
+		i = encodeVarintLogservice(dAtA, i, uint64(m.LeaseHolderID))
+		i--
+		dAtA[i] = 0x10
+	}
+	if m.Index != 0 {
+		i = encodeVarintLogservice(dAtA, i, uint64(m.Index))
+		i--
+		dAtA[i] = 0x8
+	}
+	return len(dAtA) - i, nil
+}
+
 func (m *LogRecord) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
@@ -1280,6 +1512,23 @@ func (m *LogShardInfo) Size() (n int) {
 	return n
 }
 
+func (m *LogReplicaInfo) Size() (n int) {
+	if m == nil {
+		return 0
+	}
+	var l int
+	_ = l
+	l = m.LogShardInfo.Size()
+	n += 1 + l + sovLogservice(uint64(l))
+	if m.ReplicaID != 0 {
+		n += 1 + sovLogservice(uint64(m.ReplicaID))
+	}
+	if m.XXX_unrecognized != nil {
+		n += len(m.XXX_unrecognized)
+	}
+	return n
+}
+
 func (m *LogStoreHeartbeat) Size() (n int) {
 	if m == nil {
 		return 0
@@ -1302,8 +1551,8 @@ func (m *LogStoreHeartbeat) Size() (n int) {
 	if l > 0 {
 		n += 1 + l + sovLogservice(uint64(l))
 	}
-	if len(m.Shards) > 0 {
-		for _, e := range m.Shards {
+	if len(m.Replicas) > 0 {
+		for _, e := range m.Replicas {
 			l = e.Size()
 			n += 1 + l + sovLogservice(uint64(l))
 		}
@@ -1354,6 +1603,35 @@ func (m *DNStoreHeartbeat) Size() (n int) {
 	return n
 }
 
+func (m *RSMState) Size() (n int) {
+	if m == nil {
+		return 0
+	}
+	var l int
+	_ = l
+	if m.Index != 0 {
+		n += 1 + sovLogservice(uint64(m.Index))
+	}
+	if m.LeaseHolderID != 0 {
+		n += 1 + sovLogservice(uint64(m.LeaseHolderID))
+	}
+	if m.TruncatedIndex != 0 {
+		n += 1 + sovLogservice(uint64(m.TruncatedIndex))
+	}
+	if len(m.LeaseHistory) > 0 {
+		for k, v := range m.LeaseHistory {
+			_ = k
+			_ = v
+			mapEntrySize := 1 + sovLogservice(uint64(k)) + 1 + sovLogservice(uint64(v))
+			n += mapEntrySize + 1 + sovLogservice(uint64(mapEntrySize))
+		}
+	}
+	if m.XXX_unrecognized != nil {
+		n += len(m.XXX_unrecognized)
+	}
+	return n
+}
+
 func (m *LogRecord) Size() (n int) {
 	if m == nil {
 		return 0
@@ -1711,6 +1989,109 @@ func (m *LogShardInfo) Unmarshal(dAtA []byte) error {
 	}
 	return nil
 }
+func (m *LogReplicaInfo) Unmarshal(dAtA []byte) error {
+	l := len(dAtA)
+	iNdEx := 0
+	for iNdEx < l {
+		preIndex := iNdEx
+		var wire uint64
+		for shift := uint(0); ; shift += 7 {
+			if shift >= 64 {
+				return ErrIntOverflowLogservice
+			}
+			if iNdEx >= l {
+				return io.ErrUnexpectedEOF
+			}
+			b := dAtA[iNdEx]
+			iNdEx++
+			wire |= uint64(b&0x7F) << shift
+			if b < 0x80 {
+				break
+			}
+		}
+		fieldNum := int32(wire >> 3)
+		wireType := int(wire & 0x7)
+		if wireType == 4 {
+			return fmt.Errorf("proto: LogReplicaInfo: wiretype end group for non-group")
+		}
+		if fieldNum <= 0 {
+			return fmt.Errorf("proto: LogReplicaInfo: illegal tag %d (wire type %d)", fieldNum, wire)
+		}
+		switch fieldNum {
+		case 1:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field LogShardInfo", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowLogservice
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthLogservice
+			}
+			postIndex := iNdEx + msglen
+			if postIndex < 0 {
+				return ErrInvalidLengthLogservice
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			if err := m.LogShardInfo.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
+			iNdEx = postIndex
+		case 2:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field ReplicaID", wireType)
+			}
+			m.ReplicaID = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowLogservice
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.ReplicaID |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+		default:
+			iNdEx = preIndex
+			skippy, err := skipLogservice(dAtA[iNdEx:])
+			if err != nil {
+				return err
+			}
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
+				return ErrInvalidLengthLogservice
+			}
+			if (iNdEx + skippy) > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
+			iNdEx += skippy
+		}
+	}
+
+	if iNdEx > l {
+		return io.ErrUnexpectedEOF
+	}
+	return nil
+}
 func (m *LogStoreHeartbeat) Unmarshal(dAtA []byte) error {
 	l := len(dAtA)
 	iNdEx := 0
@@ -1870,7 +2251,7 @@ func (m *LogStoreHeartbeat) Unmarshal(dAtA []byte) error {
 			iNdEx = postIndex
 		case 5:
 			if wireType != 2 {
-				return fmt.Errorf("proto: wrong wireType = %d for field Shards", wireType)
+				return fmt.Errorf("proto: wrong wireType = %d for field Replicas", wireType)
 			}
 			var msglen int
 			for shift := uint(0); ; shift += 7 {
@@ -1897,8 +2278,8 @@ func (m *LogStoreHeartbeat) Unmarshal(dAtA []byte) error {
 			if postIndex > l {
 				return io.ErrUnexpectedEOF
 			}
-			m.Shards = append(m.Shards, LogShardInfo{})
-			if err := m.Shards[len(m.Shards)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+			m.Replicas = append(m.Replicas, LogReplicaInfo{})
+			if err := m.Replicas[len(m.Replicas)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
 				return err
 			}
 			iNdEx = postIndex
@@ -2130,6 +2511,213 @@ func (m *DNStoreHeartbeat) Unmarshal(dAtA []byte) error {
 	}
 	return nil
 }
+func (m *RSMState) Unmarshal(dAtA []byte) error {
+	l := len(dAtA)
+	iNdEx := 0
+	for iNdEx < l {
+		preIndex := iNdEx
+		var wire uint64
+		for shift := uint(0); ; shift += 7 {
+			if shift >= 64 {
+				return ErrIntOverflowLogservice
+			}
+			if iNdEx >= l {
+				return io.ErrUnexpectedEOF
+			}
+			b := dAtA[iNdEx]
+			iNdEx++
+			wire |= uint64(b&0x7F) << shift
+			if b < 0x80 {
+				break
+			}
+		}
+		fieldNum := int32(wire >> 3)
+		wireType := int(wire & 0x7)
+		if wireType == 4 {
+			return fmt.Errorf("proto: RSMState: wiretype end group for non-group")
+		}
+		if fieldNum <= 0 {
+			return fmt.Errorf("proto: RSMState: illegal tag %d (wire type %d)", fieldNum, wire)
+		}
+		switch fieldNum {
+		case 1:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Index", wireType)
+			}
+			m.Index = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowLogservice
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.Index |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+		case 2:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field LeaseHolderID", wireType)
+			}
+			m.LeaseHolderID = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowLogservice
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.LeaseHolderID |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+		case 3:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field TruncatedIndex", wireType)
+			}
+			m.TruncatedIndex = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowLogservice
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.TruncatedIndex |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+		case 4:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field LeaseHistory", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowLogservice
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthLogservice
+			}
+			postIndex := iNdEx + msglen
+			if postIndex < 0 {
+				return ErrInvalidLengthLogservice
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			if m.LeaseHistory == nil {
+				m.LeaseHistory = make(map[uint64]uint64)
+			}
+			var mapkey uint64
+			var mapvalue uint64
+			for iNdEx < postIndex {
+				entryPreIndex := iNdEx
+				var wire uint64
+				for shift := uint(0); ; shift += 7 {
+					if shift >= 64 {
+						return ErrIntOverflowLogservice
+					}
+					if iNdEx >= l {
+						return io.ErrUnexpectedEOF
+					}
+					b := dAtA[iNdEx]
+					iNdEx++
+					wire |= uint64(b&0x7F) << shift
+					if b < 0x80 {
+						break
+					}
+				}
+				fieldNum := int32(wire >> 3)
+				if fieldNum == 1 {
+					for shift := uint(0); ; shift += 7 {
+						if shift >= 64 {
+							return ErrIntOverflowLogservice
+						}
+						if iNdEx >= l {
+							return io.ErrUnexpectedEOF
+						}
+						b := dAtA[iNdEx]
+						iNdEx++
+						mapkey |= uint64(b&0x7F) << shift
+						if b < 0x80 {
+							break
+						}
+					}
+				} else if fieldNum == 2 {
+					for shift := uint(0); ; shift += 7 {
+						if shift >= 64 {
+							return ErrIntOverflowLogservice
+						}
+						if iNdEx >= l {
+							return io.ErrUnexpectedEOF
+						}
+						b := dAtA[iNdEx]
+						iNdEx++
+						mapvalue |= uint64(b&0x7F) << shift
+						if b < 0x80 {
+							break
+						}
+					}
+				} else {
+					iNdEx = entryPreIndex
+					skippy, err := skipLogservice(dAtA[iNdEx:])
+					if err != nil {
+						return err
+					}
+					if (skippy < 0) || (iNdEx+skippy) < 0 {
+						return ErrInvalidLengthLogservice
+					}
+					if (iNdEx + skippy) > postIndex {
+						return io.ErrUnexpectedEOF
+					}
+					iNdEx += skippy
+				}
+			}
+			m.LeaseHistory[mapkey] = mapvalue
+			iNdEx = postIndex
+		default:
+			iNdEx = preIndex
+			skippy, err := skipLogservice(dAtA[iNdEx:])
+			if err != nil {
+				return err
+			}
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
+				return ErrInvalidLengthLogservice
+			}
+			if (iNdEx + skippy) > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
+			iNdEx += skippy
+		}
+	}
+
+	if iNdEx > l {
+		return io.ErrUnexpectedEOF
+	}
+	return nil
+}
 func (m *LogRecord) Unmarshal(dAtA []byte) error {
 	l := len(dAtA)
 	iNdEx := 0
diff --git a/proto/hakeeper.proto b/proto/hakeeper.proto
index 7b972e39c..4626fb4c2 100644
--- a/proto/hakeeper.proto
+++ b/proto/hakeeper.proto
@@ -15,6 +15,8 @@ syntax = "proto3";
 package hakeeper;
 
 import "github.com/gogo/protobuf/gogoproto/gogo.proto";
+import "logservice.proto";
+import "metadata.proto";
 
 option (gogoproto.marshaler_all)           = true;
 option (gogoproto.sizer_all)               = true;
@@ -59,3 +61,65 @@ message CommandBatch {
   uint64 Term = 1;
   repeated ScheduleCommand Commands = 2 [(gogoproto.nullable) = false]; 
 }
+
+// DNStoreInfo contins information on a list of shards.
+message DNStoreInfo {
+  uint64 Tick = 1;
+  repeated logservice.DNShardInfo Shards = 2 [(gogoproto.nullable) = false];
+}
+
+// DNState contains all DN details known to the HAKeeper.
+message DNState {
+  // Stores is keyed by DN store UUID, it contains details found on each DN
+  // store. Each DNStoreInfo reflects what was last reported by each DN store.
+  map<string, DNStoreInfo> Stores = 1 [(gogoproto.nullable) = false];
+}
+
+// ClusterInfo provides a global view of all shards in the cluster. It
+// describes the logical sharding of the system, rather than physical
+// distribution of all replicas that belong to those shards.
+message ClusterInfo {
+  repeated metadata.DNShardRecord DNShards = 1 [(gogoproto.nullable) = false];
+  repeated metadata.LogShardRecord LogShards = 2 [(gogoproto.nullable) = false];
+}
+
+// LogStoreInfo contains information of all replicas found on a Log store.
+message LogStoreInfo {
+  uint64 Tick = 1;
+  string RaftAddress = 2;
+  string ServiceAddress = 3;
+  string GossipAddress = 4;
+  repeated logservice.LogReplicaInfo Replicas = 5 [(gogoproto.nullable) = false];
+}
+
+message LogState {
+  // Shards is keyed by ShardID, it contains details aggregated from all Log
+  // stores. Each pb.LogShardInfo here contains data aggregated from
+  // different replicas and thus reflect a more accurate description on each
+  // shard.
+  map<uint64, logservice.LogShardInfo> Shards = 1 [(gogoproto.nullable) = false];
+  // Stores is keyed by log store UUID, it contains details found on each store.
+  // Each LogStoreInfo here reflects what was last reported by each Log store.
+  map<string, LogStoreInfo> Stores = 2 [(gogoproto.nullable) = false];
+}
+
+// HAKeeperState contains all HAKeeper state required for making schedule
+// commands.
+message HAKeeperState {
+  uint64 Tick = 1;
+  ClusterInfo ClusterInfo = 2 [(gogoproto.nullable) = false];
+  DNState DNState = 3 [(gogoproto.nullable) = false];
+  LogState LogState = 4 [(gogoproto.nullable) = false];
+}
+
+// RSMState contains state maintained by HAKeeper's RSM.  
+message RSMState {
+  uint64 Tick = 1;
+  uint64 NextID = 2;
+  uint64 Term = 3;
+  map<string, CommandBatch> ScheduleCommands = 4 [(gogoproto.nullable) = false];
+  map<string, uint64> LogShards = 5;
+  DNState DNState = 6 [(gogoproto.nullable) = false];
+  LogState LogState = 7 [(gogoproto.nullable) = false];
+  ClusterInfo ClusterInfo = 8 [(gogoproto.nullable) = false];
+}
diff --git a/proto/logservice.proto b/proto/logservice.proto
index 7cf7d2fb6..21708e180 100644
--- a/proto/logservice.proto
+++ b/proto/logservice.proto
@@ -13,11 +13,13 @@
 
 syntax = "proto3";
 package logservice;
+option go_package = "github.com/matrixorigin/matrixone/pkg/pb/logservice";
 
 import "github.com/gogo/protobuf/gogoproto/gogo.proto";
 
-// LogShardInfo contains information of a launched shard.  
+// LogShardInfo contains information a log shard.
 message LogShardInfo {
+  // ShardID is the ID of a Log shard.
   uint64 ShardID               = 1;
   // Replicas is a map of ReplicaID to LogStore UUID, it describe the member
   // replicas of the shard at the given Epoch. 
@@ -34,6 +36,13 @@ message LogShardInfo {
   // TODO: per shard stats like CPU/memory/network usage can be added here
 };
 
+// LogReplicaInfo contains information of a log replica.
+message LogReplicaInfo {
+  LogShardInfo LogShardInfo = 1 [(gogoproto.nullable) = false, (gogoproto.embed) = true];
+  // ReplciaID is the ID of a replica within the Log shard. 
+  uint64 ReplicaID = 2;
+}
+
 // LogStoreHeartbeat is the periodic message sent to the HAKeeper by Log Stores.
 message LogStoreHeartbeat {
   // UUID is the uuid of the Log Store.
@@ -43,11 +52,11 @@ message LogStoreHeartbeat {
   string ServiceAddress = 3;
   string GossipAddress  = 4;
 
-  // Shards is a list of LogShardInfo instances collected on the specified
+  // Replicas is a list of LogReplicaInfo instances collected on the specified
   // LogStore. Details in Shards are based on the local knowledge of each
   // replica running on the current LogStore, it may not be accurate or
   // update to date due to various reasons. 
-  repeated LogShardInfo Shards = 5 [(gogoproto.nullable) = false];
+  repeated LogReplicaInfo Replicas = 5 [(gogoproto.nullable) = false];
 };
 
 // DNShardInfo contains information of a launched DN shard. 
@@ -72,6 +81,13 @@ message DNStoreHeartbeat {
   repeated DNShardInfo Shards = 2 [(gogoproto.nullable) = false];
 };
 
+message RSMState {
+  uint64 Index = 1;
+  uint64 LeaseHolderID = 2;
+  uint64 TruncatedIndex = 3;
+  map<uint64, uint64> LeaseHistory = 4; // log index -> truncate index
+}
+
 enum MethodType {
   CREATE       = 0;
   DESTROY      = 1;


Test patch:
diff --git a/pkg/hakeeper/checkers/dnservice/check_test.go b/pkg/hakeeper/checkers/dnservice/check_test.go
index c76ffcb6e..859a3beeb 100644
--- a/pkg/hakeeper/checkers/dnservice/check_test.go
+++ b/pkg/hakeeper/checkers/dnservice/check_test.go
@@ -19,6 +19,7 @@ import (
 	"testing"
 
 	"github.com/matrixorigin/matrixone/pkg/hakeeper"
+	hapb "github.com/matrixorigin/matrixone/pkg/pb/hakeeper"
 	"github.com/matrixorigin/matrixone/pkg/pb/logservice"
 
 	"github.com/stretchr/testify/require"
@@ -192,8 +193,8 @@ func TestCheck(t *testing.T) {
 
 	// 1. no working dn stores
 	{
-		dnState := hakeeper.DNState{
-			Stores: map[string]hakeeper.DNStoreInfo{
+		dnState := hapb.DNState{
+			Stores: map[string]hapb.DNStoreInfo{
 				"expired1": {
 					Tick: expiredTick,
 					Shards: []logservice.DNShardInfo{
@@ -215,8 +216,8 @@ func TestCheck(t *testing.T) {
 
 	// 2. running cluster
 	{
-		dnState := hakeeper.DNState{
-			Stores: map[string]hakeeper.DNStoreInfo{
+		dnState := hapb.DNState{
+			Stores: map[string]hapb.DNStoreInfo{
 				"expired1": {
 					Tick: expiredTick,
 					Shards: []logservice.DNShardInfo{
@@ -266,6 +267,6 @@ func TestCheck(t *testing.T) {
 	}
 }
 
-func mockClusterInfo() hakeeper.ClusterInfo {
-	return hakeeper.ClusterInfo{}
+func mockClusterInfo() hapb.ClusterInfo {
+	return hapb.ClusterInfo{}
 }
diff --git a/pkg/hakeeper/checkers/dnservice/parse_test.go b/pkg/hakeeper/checkers/dnservice/parse_test.go
index bf85b2899..e90fdcbee 100644
--- a/pkg/hakeeper/checkers/dnservice/parse_test.go
+++ b/pkg/hakeeper/checkers/dnservice/parse_test.go
@@ -18,6 +18,7 @@ import (
 	"testing"
 
 	"github.com/matrixorigin/matrixone/pkg/hakeeper"
+	hapb "github.com/matrixorigin/matrixone/pkg/pb/hakeeper"
 	"github.com/matrixorigin/matrixone/pkg/pb/logservice"
 
 	"github.com/stretchr/testify/require"
@@ -30,8 +31,8 @@ func TestParseDNState(t *testing.T) {
 
 	// 1. no working dn stores
 	{
-		dnState := hakeeper.DNState{
-			Stores: map[string]hakeeper.DNStoreInfo{
+		dnState := hapb.DNState{
+			Stores: map[string]hapb.DNStoreInfo{
 				"expired1": {
 					Tick: expiredTick,
 					Shards: []logservice.DNShardInfo{
@@ -69,8 +70,8 @@ func TestParseDNState(t *testing.T) {
 
 	// 2. verbose running shard replica
 	{
-		dnState := hakeeper.DNState{
-			Stores: map[string]hakeeper.DNStoreInfo{
+		dnState := hapb.DNState{
+			Stores: map[string]hapb.DNStoreInfo{
 				"expired1": {
 					Tick: expiredTick,
 					Shards: []logservice.DNShardInfo{
diff --git a/pkg/hakeeper/checkers/logservice/checker_test.go b/pkg/hakeeper/checkers/logservice/checker_test.go
index 45e21bda3..c82014bc9 100644
--- a/pkg/hakeeper/checkers/logservice/checker_test.go
+++ b/pkg/hakeeper/checkers/logservice/checker_test.go
@@ -15,7 +15,7 @@ package logservice
 
 import (
 	"fmt"
-	"github.com/matrixorigin/matrixone/pkg/hakeeper"
+	"github.com/matrixorigin/matrixone/pkg/pb/hakeeper"
 	"github.com/matrixorigin/matrixone/pkg/pb/logservice"
 	"github.com/matrixorigin/matrixone/pkg/pb/metadata"
 	"reflect"
@@ -49,28 +49,31 @@ func TestCollect(t *testing.T) {
 					}},
 				Stores: map[string]hakeeper.LogStoreInfo{
 					"a": {Tick: 0, RaftAddress: "", ServiceAddress: "", GossipAddress: "",
-						Shards: []logservice.LogShardInfo{{
-							ShardID:  1,
-							Replicas: map[uint64]string{1: "a", 2: "b", 3: "c"},
-							Epoch:    1,
-							LeaderID: 0,
-							Term:     0,
-						}}},
+						Replicas: []logservice.LogReplicaInfo{{
+							LogShardInfo: logservice.LogShardInfo{
+								ShardID:  1,
+								Replicas: map[uint64]string{1: "a", 2: "b", 3: "c"},
+								Epoch:    1,
+								LeaderID: 0,
+								Term:     0,
+							}}}},
 					"b": {Tick: 0, RaftAddress: "", ServiceAddress: "", GossipAddress: "",
-						Shards: []logservice.LogShardInfo{{
-							ShardID:  1,
-							Replicas: map[uint64]string{1: "a", 2: "b", 3: "c"},
-							Epoch:    1,
-							LeaderID: 0,
-							Term:     0,
-						}}},
+						Replicas: []logservice.LogReplicaInfo{{
+							LogShardInfo: logservice.LogShardInfo{
+								ShardID:  1,
+								Replicas: map[uint64]string{1: "a", 2: "b", 3: "c"},
+								Epoch:    1,
+								LeaderID: 0,
+								Term:     0,
+							}}}},
 					"c": {Tick: 0, RaftAddress: "", ServiceAddress: "", GossipAddress: "",
-						Shards: []logservice.LogShardInfo{{
-							ShardID:  1,
-							Replicas: map[uint64]string{1: "a", 2: "b", 3: "c"},
-							Epoch:    1,
-							LeaderID: 0,
-							Term:     0}}}},
+						Replicas: []logservice.LogReplicaInfo{{
+							LogShardInfo: logservice.LogShardInfo{
+								ShardID:  1,
+								Replicas: map[uint64]string{1: "a", 2: "b", 3: "c"},
+								Epoch:    1,
+								LeaderID: 0,
+								Term:     0}}}}},
 			}, tick: 10, expected: stats{toRemove: map[uint64][]replica{}, toAdd: map[uint64]int{}}},
 		{
 			desc: "Shard 1 has only 2 replicas, which is expected as 3.",
@@ -89,19 +92,21 @@ func TestCollect(t *testing.T) {
 					Term:     0}},
 				Stores: map[string]hakeeper.LogStoreInfo{
 					"a": {Tick: 0, RaftAddress: "", ServiceAddress: "", GossipAddress: "",
-						Shards: []logservice.LogShardInfo{{
-							ShardID:  1,
-							Replicas: map[uint64]string{1: "a", 2: "b", 3: "c"},
-							Epoch:    1,
-							LeaderID: 0,
-							Term:     0}}},
+						Replicas: []logservice.LogReplicaInfo{{
+							LogShardInfo: logservice.LogShardInfo{
+								ShardID:  1,
+								Replicas: map[uint64]string{1: "a", 2: "b", 3: "c"},
+								Epoch:    1,
+								LeaderID: 0,
+								Term:     0}}}},
 					"b": {Tick: 0, RaftAddress: "", ServiceAddress: "", GossipAddress: "",
-						Shards: []logservice.LogShardInfo{{
-							ShardID:  1,
-							Replicas: map[uint64]string{1: "a", 2: "b", 3: "c"},
-							Epoch:    1,
-							LeaderID: 0,
-							Term:     0}}},
+						Replicas: []logservice.LogReplicaInfo{{
+							LogShardInfo: logservice.LogShardInfo{
+								ShardID:  1,
+								Replicas: map[uint64]string{1: "a", 2: "b", 3: "c"},
+								Epoch:    1,
+								LeaderID: 0,
+								Term:     0}}}},
 				},
 			}, tick: 10, expected: stats{toRemove: map[uint64][]replica{}, toAdd: map[uint64]int{1: 1}}},
 		{
@@ -116,21 +121,23 @@ func TestCollect(t *testing.T) {
 						Term:     0}},
 				Stores: map[string]hakeeper.LogStoreInfo{
 					"a": {Tick: 0, RaftAddress: "", ServiceAddress: "", GossipAddress: "",
-						Shards: []logservice.LogShardInfo{{
-							ShardID:  1,
-							Replicas: map[uint64]string{1: "a", 2: "b", 3: "c"},
-							Epoch:    1,
-							LeaderID: 0,
-							Term:     0}}},
+						Replicas: []logservice.LogReplicaInfo{{
+							LogShardInfo: logservice.LogShardInfo{
+								ShardID:  1,
+								Replicas: map[uint64]string{1: "a", 2: "b", 3: "c"},
+								Epoch:    1,
+								LeaderID: 0,
+								Term:     0}}}},
 					"b": {Tick: 0, RaftAddress: "", ServiceAddress: "", GossipAddress: "",
-						Shards: []logservice.LogShardInfo{{
-							ShardID:  1,
-							Replicas: map[uint64]string{1: "a", 2: "b", 3: "c"},
-							Epoch:    1,
-							LeaderID: 0,
-							Term:     0}}},
+						Replicas: []logservice.LogReplicaInfo{{
+							LogShardInfo: logservice.LogShardInfo{
+								ShardID:  1,
+								Replicas: map[uint64]string{1: "a", 2: "b", 3: "c"},
+								Epoch:    1,
+								LeaderID: 0,
+								Term:     0}}}},
 					"c": {Tick: 0, RaftAddress: "", ServiceAddress: "", GossipAddress: "",
-						Shards: []logservice.LogShardInfo{}},
+						Replicas: []logservice.LogReplicaInfo{}},
 				},
 			}, tick: 10,
 			expected: stats{toStart: []replica{{"c", 1, 0, 3}},
@@ -147,33 +154,37 @@ func TestCollect(t *testing.T) {
 						Term:     0}},
 				Stores: map[string]hakeeper.LogStoreInfo{
 					"a": {Tick: 0, RaftAddress: "", ServiceAddress: "", GossipAddress: "",
-						Shards: []logservice.LogShardInfo{{
-							ShardID:  1,
-							Replicas: map[uint64]string{1: "a", 2: "b", 3: "c"},
-							Epoch:    1,
-							LeaderID: 0,
-							Term:     0}}},
+						Replicas: []logservice.LogReplicaInfo{{
+							LogShardInfo: logservice.LogShardInfo{
+								ShardID:  1,
+								Replicas: map[uint64]string{1: "a", 2: "b", 3: "c"},
+								Epoch:    1,
+								LeaderID: 0,
+								Term:     0}}}},
 					"b": {Tick: 0, RaftAddress: "", ServiceAddress: "", GossipAddress: "",
-						Shards: []logservice.LogShardInfo{{
-							ShardID:  1,
-							Replicas: map[uint64]string{1: "a", 2: "b", 3: "c"},
-							Epoch:    1,
-							LeaderID: 0,
-							Term:     0}}},
+						Replicas: []logservice.LogReplicaInfo{{
+							LogShardInfo: logservice.LogShardInfo{
+								ShardID:  1,
+								Replicas: map[uint64]string{1: "a", 2: "b", 3: "c"},
+								Epoch:    1,
+								LeaderID: 0,
+								Term:     0}}}},
 					"c": {Tick: 0, RaftAddress: "", ServiceAddress: "", GossipAddress: "",
-						Shards: []logservice.LogShardInfo{{
-							ShardID:  1,
-							Replicas: map[uint64]string{1: "a", 2: "b", 3: "c"},
-							Epoch:    1,
-							LeaderID: 0,
-							Term:     0}}},
+						Replicas: []logservice.LogReplicaInfo{{
+							LogShardInfo: logservice.LogShardInfo{
+								ShardID:  1,
+								Replicas: map[uint64]string{1: "a", 2: "b", 3: "c"},
+								Epoch:    1,
+								LeaderID: 0,
+								Term:     0}}}},
 					"d": {Tick: 0, RaftAddress: "", ServiceAddress: "", GossipAddress: "",
-						Shards: []logservice.LogShardInfo{{
-							ShardID:  1,
-							Replicas: map[uint64]string{1: "a", 2: "b", 4: "d"},
-							Epoch:    0,
-							LeaderID: 0,
-							Term:     0}}},
+						Replicas: []logservice.LogReplicaInfo{{
+							LogShardInfo: logservice.LogShardInfo{
+								ShardID:  1,
+								Replicas: map[uint64]string{1: "a", 2: "b", 4: "d"},
+								Epoch:    0,
+								LeaderID: 0,
+								Term:     0}}}},
 				},
 			}, tick: 10,
 			expected: stats{toStop: []replica{{"d", 1, 0, 0}},
@@ -196,33 +207,37 @@ func TestCollect(t *testing.T) {
 						Term:     0}},
 				Stores: map[string]hakeeper.LogStoreInfo{
 					"a": {Tick: 0, RaftAddress: "", ServiceAddress: "", GossipAddress: "",
-						Shards: []logservice.LogShardInfo{{
-							ShardID:  1,
-							Replicas: map[uint64]string{1: "a", 2: "b", 3: "c", 4: "d"},
-							Epoch:    1,
-							LeaderID: 0,
-							Term:     0}}},
+						Replicas: []logservice.LogReplicaInfo{{
+							LogShardInfo: logservice.LogShardInfo{
+								ShardID:  1,
+								Replicas: map[uint64]string{1: "a", 2: "b", 3: "c", 4: "d"},
+								Epoch:    1,
+								LeaderID: 0,
+								Term:     0}}}},
 					"b": {Tick: 0, RaftAddress: "", ServiceAddress: "", GossipAddress: "",
-						Shards: []logservice.LogShardInfo{{
-							ShardID:  1,
-							Replicas: map[uint64]string{1: "a", 2: "b", 3: "c", 4: "d"},
-							Epoch:    1,
-							LeaderID: 0,
-							Term:     0}}},
+						Replicas: []logservice.LogReplicaInfo{{
+							LogShardInfo: logservice.LogShardInfo{
+								ShardID:  1,
+								Replicas: map[uint64]string{1: "a", 2: "b", 3: "c", 4: "d"},
+								Epoch:    1,
+								LeaderID: 0,
+								Term:     0}}}},
 					"c": {Tick: 0, RaftAddress: "", ServiceAddress: "", GossipAddress: "",
-						Shards: []logservice.LogShardInfo{{
-							ShardID:  1,
-							Replicas: map[uint64]string{1: "a", 2: "b", 3: "c", 4: "d"},
-							Epoch:    1,
-							LeaderID: 0,
-							Term:     0}}},
+						Replicas: []logservice.LogReplicaInfo{{
+							LogShardInfo: logservice.LogShardInfo{
+								ShardID:  1,
+								Replicas: map[uint64]string{1: "a", 2: "b", 3: "c", 4: "d"},
+								Epoch:    1,
+								LeaderID: 0,
+								Term:     0}}}},
 					"d": {Tick: 0, RaftAddress: "", ServiceAddress: "", GossipAddress: "",
-						Shards: []logservice.LogShardInfo{{
-							ShardID:  1,
-							Replicas: map[uint64]string{1: "a", 2: "b", 3: "c", 4: "d"},
-							Epoch:    1,
-							LeaderID: 0,
-							Term:     0}}},
+						Replicas: []logservice.LogReplicaInfo{{
+							LogShardInfo: logservice.LogShardInfo{
+								ShardID:  1,
+								Replicas: map[uint64]string{1: "a", 2: "b", 3: "c", 4: "d"},
+								Epoch:    1,
+								LeaderID: 0,
+								Term:     0}}}},
 				},
 			}, tick: 10,
 			expected: stats{toRemove: map[uint64][]replica{1: {{"a", 1, 0, 1}}},
@@ -239,26 +254,29 @@ func TestCollect(t *testing.T) {
 						Term:     0}},
 				Stores: map[string]hakeeper.LogStoreInfo{
 					"a": {Tick: 0, RaftAddress: "", ServiceAddress: "", GossipAddress: "",
-						Shards: []logservice.LogShardInfo{{
-							ShardID:  1,
-							Replicas: map[uint64]string{1: "a", 2: "b", 3: "c"},
-							Epoch:    1,
-							LeaderID: 0,
-							Term:     0}}},
+						Replicas: []logservice.LogReplicaInfo{{
+							LogShardInfo: logservice.LogShardInfo{
+								ShardID:  1,
+								Replicas: map[uint64]string{1: "a", 2: "b", 3: "c"},
+								Epoch:    1,
+								LeaderID: 0,
+								Term:     0}}}},
 					"b": {Tick: 999999999, RaftAddress: "", ServiceAddress: "", GossipAddress: "",
-						Shards: []logservice.LogShardInfo{{
-							ShardID:  1,
-							Replicas: map[uint64]string{1: "a", 2: "b", 3: "c"},
-							Epoch:    1,
-							LeaderID: 0,
-							Term:     0}}},
+						Replicas: []logservice.LogReplicaInfo{{
+							LogShardInfo: logservice.LogShardInfo{
+								ShardID:  1,
+								Replicas: map[uint64]string{1: "a", 2: "b", 3: "c"},
+								Epoch:    1,
+								LeaderID: 0,
+								Term:     0}}}},
 					"c": {Tick: 999999999, RaftAddress: "", ServiceAddress: "", GossipAddress: "",
-						Shards: []logservice.LogShardInfo{{
-							ShardID:  1,
-							Replicas: map[uint64]string{1: "a", 2: "b", 3: "c"},
-							Epoch:    1,
-							LeaderID: 0,
-							Term:     0}}}},
+						Replicas: []logservice.LogReplicaInfo{{
+							LogShardInfo: logservice.LogShardInfo{
+								ShardID:  1,
+								Replicas: map[uint64]string{1: "a", 2: "b", 3: "c"},
+								Epoch:    1,
+								LeaderID: 0,
+								Term:     0}}}}},
 			}, tick: 999999999, expected: stats{
 				toRemove: map[uint64][]replica{1: {{"a", 1, 1, 1}}},
 				toAdd:    map[uint64]int{}}},
diff --git a/pkg/hakeeper/operator/operator_test.go b/pkg/hakeeper/operator/operator_test.go
index e8230d0b5..f0aa82c1b 100644
--- a/pkg/hakeeper/operator/operator_test.go
+++ b/pkg/hakeeper/operator/operator_test.go
@@ -23,7 +23,7 @@ package operator
 import (
 	"testing"
 
-	"github.com/matrixorigin/matrixone/pkg/hakeeper"
+	"github.com/matrixorigin/matrixone/pkg/pb/hakeeper"
 	"github.com/matrixorigin/matrixone/pkg/pb/logservice"
 	"github.com/stretchr/testify/assert"
 )
diff --git a/pkg/hakeeper/rsm_test.go b/pkg/hakeeper/rsm_test.go
index 58b9c0eff..a0996b182 100644
--- a/pkg/hakeeper/rsm_test.go
+++ b/pkg/hakeeper/rsm_test.go
@@ -28,9 +28,9 @@ import (
 
 func TestAssignID(t *testing.T) {
 	tsm := NewStateMachine(0, 1).(*stateMachine)
-	assert.Equal(t, uint64(0), tsm.NextID)
+	assert.Equal(t, uint64(0), tsm.state.NextID)
 	assert.Equal(t, uint64(1), tsm.assignID())
-	assert.Equal(t, uint64(1), tsm.NextID)
+	assert.Equal(t, uint64(1), tsm.state.NextID)
 }
 
 func TestCreateLogShardCmd(t *testing.T) {
@@ -54,29 +54,29 @@ func TestHAKeeperStateMachineCanBeCreated(t *testing.T) {
 func TestHAKeeperStateMachineSnapshot(t *testing.T) {
 	tsm1 := NewStateMachine(0, 1).(*stateMachine)
 	tsm2 := NewStateMachine(0, 2).(*stateMachine)
-	tsm1.NextID = 12345
-	tsm1.LogShards["test1"] = 23456
-	tsm1.LogShards["test2"] = 34567
+	tsm1.state.NextID = 12345
+	tsm1.state.LogShards["test1"] = 23456
+	tsm1.state.LogShards["test2"] = 34567
 
 	buf := bytes.NewBuffer(nil)
 	assert.Nil(t, tsm1.SaveSnapshot(buf, nil, nil))
 	assert.Nil(t, tsm2.RecoverFromSnapshot(buf, nil, nil))
-	assert.Equal(t, tsm1.NextID, tsm2.NextID)
-	assert.Equal(t, tsm1.LogShards, tsm2.LogShards)
+	assert.Equal(t, tsm1.state.NextID, tsm2.state.NextID)
+	assert.Equal(t, tsm1.state.LogShards, tsm2.state.LogShards)
 	assert.True(t, tsm1.replicaID != tsm2.replicaID)
 }
 
 func TestHAKeeperLogShardCanBeCreated(t *testing.T) {
 	cmd := getCreateLogShardCmd("test1")
 	tsm1 := NewStateMachine(0, 1).(*stateMachine)
-	tsm1.NextID = 100
+	tsm1.state.NextID = 100
 
 	result, err := tsm1.Update(sm.Entry{Cmd: cmd})
 	assert.Nil(t, err)
 	assert.Equal(t, sm.Result{Value: 101}, result)
-	assert.Equal(t, uint64(101), tsm1.NextID)
+	assert.Equal(t, uint64(101), tsm1.state.NextID)
 
-	tsm1.NextID = 200
+	tsm1.state.NextID = 200
 	result, err = tsm1.Update(sm.Entry{Cmd: cmd})
 	assert.Nil(t, err)
 	data := make([]byte, 8)
@@ -87,7 +87,7 @@ func TestHAKeeperLogShardCanBeCreated(t *testing.T) {
 func TestHAKeeperQueryLogShardID(t *testing.T) {
 	cmd := getCreateLogShardCmd("test1")
 	tsm1 := NewStateMachine(0, 1).(*stateMachine)
-	tsm1.NextID = 100
+	tsm1.state.NextID = 100
 	result, err := tsm1.Update(sm.Entry{Cmd: cmd})
 	assert.Nil(t, err)
 	assert.Equal(t, sm.Result{Value: 101}, result)
@@ -97,7 +97,6 @@ func TestHAKeeperQueryLogShardID(t *testing.T) {
 	assert.NoError(t, err)
 	r1, ok := r.(*logShardIDQueryResult)
 	assert.True(t, ok)
-	assert.True(t, r1.found)
 	assert.Equal(t, uint64(101), r1.id)
 
 	q2 := &logShardIDQuery{name: "test2"}
@@ -105,7 +104,7 @@ func TestHAKeeperQueryLogShardID(t *testing.T) {
 	assert.NoError(t, err)
 	r2, ok := r.(*logShardIDQueryResult)
 	assert.True(t, ok)
-	assert.False(t, r2.found)
+	assert.Equal(t, uint64(0), r2.id)
 }
 
 func TestHAKeeperCanBeClosed(t *testing.T) {
@@ -115,13 +114,13 @@ func TestHAKeeperCanBeClosed(t *testing.T) {
 
 func TestHAKeeperTick(t *testing.T) {
 	tsm1 := NewStateMachine(0, 1).(*stateMachine)
-	assert.Equal(t, uint64(0), tsm1.Tick)
+	assert.Equal(t, uint64(0), tsm1.state.Tick)
 	cmd := GetTickCmd()
 	_, err := tsm1.Update(sm.Entry{Cmd: cmd})
 	assert.NoError(t, err)
 	_, err = tsm1.Update(sm.Entry{Cmd: cmd})
 	assert.NoError(t, err)
-	assert.Equal(t, uint64(2), tsm1.Tick)
+	assert.Equal(t, uint64(2), tsm1.state.Tick)
 }
 
 func TestHandleLogHeartbeat(t *testing.T) {
@@ -139,26 +138,30 @@ func TestHandleLogHeartbeat(t *testing.T) {
 		RaftAddress:    "localhost:9090",
 		ServiceAddress: "localhost:9091",
 		GossipAddress:  "localhost:9092",
-		Shards: []pb.LogShardInfo{
+		Replicas: []pb.LogReplicaInfo{
 			{
-				ShardID: 100,
-				Replicas: map[uint64]string{
-					200: "localhost:8000",
-					300: "localhost:9000",
+				LogShardInfo: pb.LogShardInfo{
+					ShardID: 100,
+					Replicas: map[uint64]string{
+						200: "localhost:8000",
+						300: "localhost:9000",
+					},
+					Epoch:    200,
+					LeaderID: 200,
+					Term:     10,
 				},
-				Epoch:    200,
-				LeaderID: 200,
-				Term:     10,
 			},
 			{
-				ShardID: 101,
-				Replicas: map[uint64]string{
-					201: "localhost:8000",
-					301: "localhost:9000",
+				LogShardInfo: pb.LogShardInfo{
+					ShardID: 101,
+					Replicas: map[uint64]string{
+						201: "localhost:8000",
+						301: "localhost:9000",
+					},
+					Epoch:    202,
+					LeaderID: 201,
+					Term:     30,
 				},
-				Epoch:    202,
-				LeaderID: 201,
-				Term:     30,
 			},
 		},
 	}
@@ -167,7 +170,7 @@ func TestHandleLogHeartbeat(t *testing.T) {
 	cmd = GetLogStoreHeartbeatCmd(data)
 	_, err = tsm1.Update(sm.Entry{Cmd: cmd})
 	assert.NoError(t, err)
-	s := tsm1.LogState
+	s := tsm1.state.LogState
 	assert.Equal(t, 1, len(s.Stores))
 	lsinfo, ok := s.Stores[hb.UUID]
 	require.True(t, ok)
@@ -175,8 +178,8 @@ func TestHandleLogHeartbeat(t *testing.T) {
 	assert.Equal(t, hb.RaftAddress, lsinfo.RaftAddress)
 	assert.Equal(t, hb.ServiceAddress, lsinfo.ServiceAddress)
 	assert.Equal(t, hb.GossipAddress, lsinfo.GossipAddress)
-	assert.Equal(t, 2, len(lsinfo.Shards))
-	assert.Equal(t, hb.Shards, lsinfo.Shards)
+	assert.Equal(t, 2, len(lsinfo.Replicas))
+	assert.Equal(t, hb.Replicas, lsinfo.Replicas)
 }
 
 func TestHandleDNHeartbeat(t *testing.T) {
@@ -202,7 +205,7 @@ func TestHandleDNHeartbeat(t *testing.T) {
 	cmd = GetDNStoreHeartbeatCmd(data)
 	_, err = tsm1.Update(sm.Entry{Cmd: cmd})
 	assert.NoError(t, err)
-	s := tsm1.DNState
+	s := tsm1.state.DNState
 	assert.Equal(t, 1, len(s.Stores))
 	dninfo, ok := s.Stores[hb.UUID]
 	assert.True(t, ok)
@@ -270,26 +273,26 @@ func TestUpdateScheduleCommandsCmd(t *testing.T) {
 	result, err := tsm1.Update(sm.Entry{Cmd: cmd})
 	require.NoError(t, err)
 	assert.Equal(t, sm.Result{}, result)
-	assert.Equal(t, b.Term, tsm1.term)
-	require.Equal(t, 2, len(tsm1.scheduleCommands))
-	l1, ok := tsm1.scheduleCommands["uuid1"]
+	assert.Equal(t, b.Term, tsm1.state.Term)
+	require.Equal(t, 2, len(tsm1.state.ScheduleCommands))
+	l1, ok := tsm1.state.ScheduleCommands["uuid1"]
 	assert.True(t, ok)
-	assert.Equal(t, []hapb.ScheduleCommand{sc1, sc3}, l1)
-	l2, ok := tsm1.scheduleCommands["uuid2"]
+	assert.Equal(t, hapb.CommandBatch{Commands: []hapb.ScheduleCommand{sc1, sc3}}, l1)
+	l2, ok := tsm1.state.ScheduleCommands["uuid2"]
 	assert.True(t, ok)
-	assert.Equal(t, []hapb.ScheduleCommand{sc2}, l2)
+	assert.Equal(t, hapb.CommandBatch{Commands: []hapb.ScheduleCommand{sc2}}, l2)
 
 	cmd2 := GetUpdateCommandsCmd(b.Term-1,
 		[]hapb.ScheduleCommand{sc1, sc2, sc3, sc4})
 	result, err = tsm1.Update(sm.Entry{Cmd: cmd2})
 	require.NoError(t, err)
 	assert.Equal(t, sm.Result{}, result)
-	assert.Equal(t, b.Term, tsm1.term)
-	require.Equal(t, 2, len(tsm1.scheduleCommands))
-	l1, ok = tsm1.scheduleCommands["uuid1"]
+	assert.Equal(t, b.Term, tsm1.state.Term)
+	require.Equal(t, 2, len(tsm1.state.ScheduleCommands))
+	l1, ok = tsm1.state.ScheduleCommands["uuid1"]
 	assert.True(t, ok)
-	assert.Equal(t, []hapb.ScheduleCommand{sc1, sc3}, l1)
-	l2, ok = tsm1.scheduleCommands["uuid2"]
+	assert.Equal(t, hapb.CommandBatch{Commands: []hapb.ScheduleCommand{sc1, sc3}}, l1)
+	l2, ok = tsm1.state.ScheduleCommands["uuid2"]
 	assert.True(t, ok)
-	assert.Equal(t, []hapb.ScheduleCommand{sc2}, l2)
+	assert.Equal(t, hapb.CommandBatch{Commands: []hapb.ScheduleCommand{sc2}}, l2)
 }
diff --git a/pkg/logservice/rsm_test.go b/pkg/logservice/rsm_test.go
index a39801297..d3f8ce9d3 100644
--- a/pkg/logservice/rsm_test.go
+++ b/pkg/logservice/rsm_test.go
@@ -24,9 +24,9 @@ import (
 
 func TestGetLeaseHistory(t *testing.T) {
 	tsm := newStateMachine(1, 2).(*stateMachine)
-	tsm.LeaseHistory[100] = 1000
-	tsm.LeaseHistory[200] = 2000
-	tsm.LeaseHistory[300] = 3000
+	tsm.state.LeaseHistory[100] = 1000
+	tsm.state.LeaseHistory[200] = 2000
+	tsm.state.LeaseHistory[300] = 3000
 	lease, index := tsm.getLeaseHistory(150)
 	assert.Equal(t, uint64(1000), lease)
 	assert.Equal(t, uint64(100), index)
@@ -47,34 +47,34 @@ func TestGetLeaseHistory(t *testing.T) {
 func TestTruncateLeaseHistory(t *testing.T) {
 	getSM := func() *stateMachine {
 		tsm := newStateMachine(1, 2).(*stateMachine)
-		tsm.LeaseHistory[100] = 1000
-		tsm.LeaseHistory[200] = 2000
-		tsm.LeaseHistory[300] = 3000
+		tsm.state.LeaseHistory[100] = 1000
+		tsm.state.LeaseHistory[200] = 2000
+		tsm.state.LeaseHistory[300] = 3000
 		return tsm
 	}
 
 	tsm := getSM()
 	tsm.truncateLeaseHistory(105)
-	assert.Equal(t, 3, len(tsm.LeaseHistory))
+	assert.Equal(t, 3, len(tsm.state.LeaseHistory))
 	tsm.truncateLeaseHistory(200)
-	assert.Equal(t, 3, len(tsm.LeaseHistory))
+	assert.Equal(t, 3, len(tsm.state.LeaseHistory))
 	tsm.truncateLeaseHistory(201)
-	assert.Equal(t, 2, len(tsm.LeaseHistory))
-	_, ok1 := tsm.LeaseHistory[200]
-	_, ok2 := tsm.LeaseHistory[300]
+	assert.Equal(t, 2, len(tsm.state.LeaseHistory))
+	_, ok1 := tsm.state.LeaseHistory[200]
+	_, ok2 := tsm.state.LeaseHistory[300]
 	assert.True(t, ok1)
 	assert.True(t, ok2)
 
 	tsm = getSM()
 	tsm.truncateLeaseHistory(300)
-	assert.Equal(t, 2, len(tsm.LeaseHistory))
-	_, ok := tsm.LeaseHistory[100]
+	assert.Equal(t, 2, len(tsm.state.LeaseHistory))
+	_, ok := tsm.state.LeaseHistory[100]
 	assert.False(t, ok)
 
 	tsm = getSM()
 	tsm.truncateLeaseHistory(301)
-	assert.Equal(t, 1, len(tsm.LeaseHistory))
-	_, ok = tsm.LeaseHistory[300]
+	assert.Equal(t, 1, len(tsm.state.LeaseHistory))
+	_, ok = tsm.state.LeaseHistory[300]
 	assert.True(t, ok)
 }
 
@@ -116,14 +116,14 @@ func TestStateMachineCanBeClosed(t *testing.T) {
 func TestDNLeaseHolderCanBeUpdated(t *testing.T) {
 	cmd := getSetLeaseHolderCmd(500)
 	tsm := newStateMachine(1, 2).(*stateMachine)
-	assert.Equal(t, uint64(0), tsm.LeaseHolderID)
+	assert.Equal(t, uint64(0), tsm.state.LeaseHolderID)
 	e := sm.Entry{Cmd: cmd, Index: 100}
 	result, err := tsm.Update(e)
 	assert.Equal(t, sm.Result{}, result)
 	assert.Nil(t, err)
-	assert.Equal(t, uint64(500), tsm.LeaseHolderID)
-	assert.Equal(t, 1, len(tsm.LeaseHistory))
-	v, ok := tsm.LeaseHistory[100]
+	assert.Equal(t, uint64(500), tsm.state.LeaseHolderID)
+	assert.Equal(t, 1, len(tsm.state.LeaseHistory))
+	v, ok := tsm.state.LeaseHistory[100]
 	assert.True(t, ok)
 	assert.Equal(t, uint64(500), v)
 
@@ -132,9 +132,9 @@ func TestDNLeaseHolderCanBeUpdated(t *testing.T) {
 	result, err = tsm.Update(e)
 	assert.Equal(t, sm.Result{}, result)
 	assert.Nil(t, err)
-	assert.Equal(t, uint64(1000), tsm.LeaseHolderID)
-	assert.Equal(t, 2, len(tsm.LeaseHistory))
-	v, ok = tsm.LeaseHistory[200]
+	assert.Equal(t, uint64(1000), tsm.state.LeaseHolderID)
+	assert.Equal(t, 2, len(tsm.state.LeaseHistory))
+	v, ok = tsm.state.LeaseHistory[200]
 	assert.True(t, ok)
 	assert.Equal(t, uint64(1000), v)
 
@@ -144,7 +144,7 @@ func TestDNLeaseHolderCanBeUpdated(t *testing.T) {
 	assert.Equal(t, sm.Result{}, result)
 	assert.Nil(t, err)
 	// first lease history record won't be truncated
-	assert.Equal(t, 2, len(tsm.LeaseHistory))
+	assert.Equal(t, 2, len(tsm.state.LeaseHistory))
 }
 
 func TestTruncatedIndexCanBeUpdated(t *testing.T) {
@@ -174,53 +174,53 @@ func TestStateMachineUserUpdate(t *testing.T) {
 	binaryEnc.PutUint64(cmd[headerSize:], uint64(1234))
 
 	tsm := newStateMachine(1, 2).(*stateMachine)
-	tsm.LeaseHolderID = 1234
+	tsm.state.LeaseHolderID = 1234
 	e := sm.Entry{Index: 100, Cmd: cmd}
 	result, err := tsm.Update(e)
 	assert.Nil(t, err)
 	assert.Equal(t, e.Index, result.Value)
 	assert.Nil(t, result.Data)
 
-	tsm.LeaseHolderID = 2345
+	tsm.state.LeaseHolderID = 2345
 	e = sm.Entry{Cmd: cmd}
 	result, err = tsm.Update(e)
 	assert.Nil(t, err)
 	assert.Equal(t, uint64(0), result.Value)
 	assert.NotNil(t, result.Data)
-	assert.Equal(t, tsm.LeaseHolderID, binaryEnc.Uint64(result.Data))
+	assert.Equal(t, tsm.state.LeaseHolderID, binaryEnc.Uint64(result.Data))
 }
 
 func TestStateMachineSnapshot(t *testing.T) {
 	buf := bytes.NewBuffer(nil)
 	tsm := newStateMachine(1, 2).(*stateMachine)
-	tsm.LeaseHolderID = 123456
-	tsm.TruncatedIndex = 456789
+	tsm.state.LeaseHolderID = 123456
+	tsm.state.TruncatedIndex = 456789
 	assert.Nil(t, tsm.SaveSnapshot(buf, nil, nil))
 
 	tsm2 := newStateMachine(3, 4).(*stateMachine)
 	assert.Nil(t, tsm2.RecoverFromSnapshot(buf, nil, nil))
-	assert.Equal(t, tsm.LeaseHolderID, tsm2.LeaseHolderID)
-	assert.Equal(t, tsm.TruncatedIndex, tsm2.TruncatedIndex)
+	assert.Equal(t, tsm.state.LeaseHolderID, tsm2.state.LeaseHolderID)
+	assert.Equal(t, tsm.state.TruncatedIndex, tsm2.state.TruncatedIndex)
 	assert.Equal(t, uint64(3), tsm2.shardID)
 	assert.Equal(t, uint64(4), tsm2.replicaID)
 }
 
 func TestStateMachineLookup(t *testing.T) {
 	tsm := newStateMachine(1, 2).(*stateMachine)
-	tsm.Index = 1234
-	tsm.LeaseHolderID = 123456
-	tsm.TruncatedIndex = 456789
+	tsm.state.Index = 1234
+	tsm.state.LeaseHolderID = 123456
+	tsm.state.TruncatedIndex = 456789
 	v, err := tsm.Lookup(leaseHolderIDQuery{})
 	assert.Nil(t, err)
-	assert.Equal(t, tsm.LeaseHolderID, v.(uint64))
+	assert.Equal(t, tsm.state.LeaseHolderID, v.(uint64))
 
 	v2, err := tsm.Lookup(truncatedIndexQuery{})
 	assert.Nil(t, err)
-	assert.Equal(t, tsm.TruncatedIndex, v2.(uint64))
+	assert.Equal(t, tsm.state.TruncatedIndex, v2.(uint64))
 
 	v3, err := tsm.Lookup(indexQuery{})
 	assert.Nil(t, err)
-	assert.Equal(t, tsm.Index, v3.(uint64))
+	assert.Equal(t, tsm.state.Index, v3.(uint64))
 }
 
 func TestStateMachineLookupPanicOnUnexpectedInputValue(t *testing.T) {
diff --git a/pkg/logservice/store_test.go b/pkg/logservice/store_test.go
index 975f79846..b0dcc0b6d 100644
--- a/pkg/logservice/store_test.go
+++ b/pkg/logservice/store_test.go
@@ -258,7 +258,7 @@ func TestGetHeartbeatMessage(t *testing.T) {
 
 		m := store.getHeartbeatMessage()
 		// hakeeper shard is included
-		assert.Equal(t, 3, len(m.Shards))
+		assert.Equal(t, 3, len(m.Replicas))
 	}
 	runStoreTest(t, fn)
 }
diff --git a/pkg/hakeeper/state_test.go b/pkg/pb/hakeeper/hakeeper_test.go
similarity index 64%
rename from pkg/hakeeper/state_test.go
rename to pkg/pb/hakeeper/hakeeper_test.go
index 7a0a9b1a3..da3846495 100644
--- a/pkg/hakeeper/state_test.go
+++ b/pkg/pb/hakeeper/hakeeper_test.go
@@ -20,14 +20,14 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
-	"github.com/matrixorigin/matrixone/pkg/pb/logservice"
+	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
 )
 
 func TestDNStateUpdate(t *testing.T) {
 	s := NewDNState()
-	hb := logservice.DNStoreHeartbeat{
+	hb := pb.DNStoreHeartbeat{
 		UUID: "uuid1",
-		Shards: []logservice.DNShardInfo{
+		Shards: []pb.DNShardInfo{
 			{ShardID: 1, ReplicaID: 1},
 			{ShardID: 2, ReplicaID: 1},
 			{ShardID: 3, ReplicaID: 1},
@@ -41,17 +41,17 @@ func TestDNStateUpdate(t *testing.T) {
 	require.Equal(t, 3, len(dninfo.Shards))
 	assert.Equal(t, hb.Shards, dninfo.Shards)
 
-	hb = logservice.DNStoreHeartbeat{
+	hb = pb.DNStoreHeartbeat{
 		UUID: "uuid2",
-		Shards: []logservice.DNShardInfo{
+		Shards: []pb.DNShardInfo{
 			{ShardID: 100, ReplicaID: 1},
 		},
 	}
 	s.Update(hb, 2)
 
-	hb = logservice.DNStoreHeartbeat{
+	hb = pb.DNStoreHeartbeat{
 		UUID: "uuid1",
-		Shards: []logservice.DNShardInfo{
+		Shards: []pb.DNShardInfo{
 			{ShardID: 1, ReplicaID: 1},
 			{ShardID: 3, ReplicaID: 1},
 			{ShardID: 4, ReplicaID: 1},
@@ -69,31 +69,35 @@ func TestDNStateUpdate(t *testing.T) {
 
 func TestUpdateLogStateStore(t *testing.T) {
 	s := NewLogState()
-	hb := logservice.LogStoreHeartbeat{
+	hb := pb.LogStoreHeartbeat{
 		UUID:           "uuid1",
 		RaftAddress:    "localhost:9090",
 		ServiceAddress: "localhost:9091",
 		GossipAddress:  "localhost:9092",
-		Shards: []logservice.LogShardInfo{
+		Replicas: []pb.LogReplicaInfo{
 			{
-				ShardID: 100,
-				Replicas: map[uint64]string{
-					200: "localhost:8000",
-					300: "localhost:9000",
+				LogShardInfo: pb.LogShardInfo{
+					ShardID: 100,
+					Replicas: map[uint64]string{
+						200: "localhost:8000",
+						300: "localhost:9000",
+					},
+					Epoch:    200,
+					LeaderID: 200,
+					Term:     10,
 				},
-				Epoch:    200,
-				LeaderID: 200,
-				Term:     10,
 			},
 			{
-				ShardID: 101,
-				Replicas: map[uint64]string{
-					201: "localhost:8000",
-					301: "localhost:9000",
+				LogShardInfo: pb.LogShardInfo{
+					ShardID: 101,
+					Replicas: map[uint64]string{
+						201: "localhost:8000",
+						301: "localhost:9000",
+					},
+					Epoch:    202,
+					LeaderID: 201,
+					Term:     30,
 				},
-				Epoch:    202,
-				LeaderID: 201,
-				Term:     30,
 			},
 		},
 	}
@@ -106,42 +110,46 @@ func TestUpdateLogStateStore(t *testing.T) {
 	assert.Equal(t, hb.RaftAddress, lsinfo.RaftAddress)
 	assert.Equal(t, hb.ServiceAddress, lsinfo.ServiceAddress)
 	assert.Equal(t, hb.GossipAddress, lsinfo.GossipAddress)
-	assert.Equal(t, 2, len(lsinfo.Shards))
-	assert.Equal(t, hb.Shards, lsinfo.Shards)
+	assert.Equal(t, 2, len(lsinfo.Replicas))
+	assert.Equal(t, hb.Replicas, lsinfo.Replicas)
 
 	require.Equal(t, 2, len(s.Shards))
 	shard1, ok := s.Shards[100]
 	assert.True(t, ok)
-	assert.Equal(t, hb.Shards[0], shard1)
+	assert.Equal(t, hb.Replicas[0].LogShardInfo, shard1)
 	shard2, ok := s.Shards[101]
 	assert.True(t, ok)
-	assert.Equal(t, hb.Shards[1], shard2)
+	assert.Equal(t, hb.Replicas[1].LogShardInfo, shard2)
 
-	hb2 := logservice.LogStoreHeartbeat{
+	hb2 := pb.LogStoreHeartbeat{
 		UUID:           "uuid1",
 		RaftAddress:    "localhost:9090",
 		ServiceAddress: "localhost:9091",
 		GossipAddress:  "localhost:9092",
-		Shards: []logservice.LogShardInfo{
+		Replicas: []pb.LogReplicaInfo{
 			{
-				ShardID: 100,
-				Replicas: map[uint64]string{
-					200: "localhost:8000",
-					300: "localhost:9000",
-					400: "localhost:10000",
+				LogShardInfo: pb.LogShardInfo{
+					ShardID: 100,
+					Replicas: map[uint64]string{
+						200: "localhost:8000",
+						300: "localhost:9000",
+						400: "localhost:10000",
+					},
+					Epoch:    201,
+					LeaderID: 400,
+					Term:     20,
 				},
-				Epoch:    201,
-				LeaderID: 400,
-				Term:     20,
 			},
 			{
-				ShardID: 101,
-				Replicas: map[uint64]string{
-					201: "localhost:8000",
+				LogShardInfo: pb.LogShardInfo{
+					ShardID: 101,
+					Replicas: map[uint64]string{
+						201: "localhost:8000",
+					},
+					Epoch:    200,
+					LeaderID: NoLeader,
+					Term:     100,
 				},
-				Epoch:    200,
-				LeaderID: NoLeader,
-				Term:     100,
 			},
 		},
 	}
@@ -154,15 +162,15 @@ func TestUpdateLogStateStore(t *testing.T) {
 	assert.Equal(t, hb2.RaftAddress, lsinfo.RaftAddress)
 	assert.Equal(t, hb2.ServiceAddress, lsinfo.ServiceAddress)
 	assert.Equal(t, hb2.GossipAddress, lsinfo.GossipAddress)
-	assert.Equal(t, 2, len(lsinfo.Shards))
-	assert.Equal(t, hb2.Shards, lsinfo.Shards)
+	assert.Equal(t, 2, len(lsinfo.Replicas))
+	assert.Equal(t, hb2.Replicas, lsinfo.Replicas)
 
 	require.Equal(t, 2, len(s.Shards))
 	shard1, ok = s.Shards[100]
 	assert.True(t, ok)
-	assert.Equal(t, hb2.Shards[0], shard1)
+	assert.Equal(t, hb2.Replicas[0].LogShardInfo, shard1)
 	shard2, ok = s.Shards[101]
 	assert.True(t, ok)
 	// shard2 didn't change to hb2.Shard[1]
-	assert.Equal(t, hb.Shards[1], shard2)
+	assert.Equal(t, hb.Replicas[1].LogShardInfo, shard2)
 }