Enhancement: persist commit index in LogStore to accelerate recovery

api.go

@@ -217,6 +217,10 @@ type Raft struct {
 	// preVoteDisabled control if the pre-vote feature is activated,
 	// prevote feature is disabled if set to true.
 	preVoteDisabled bool
+
+	// fastRecovery is used to enable fast recovery mode
+	// fast recovery mode is disabled if set to false.
+	fastRecovery bool
 }
 
 // BootstrapCluster initializes a server's storage with the given cluster
@@ -566,6 +570,7 @@ func NewRaft(conf *Config, fsm FSM, logs LogStore, stable StableStore, snaps Sna
 		followerNotifyCh:      make(chan struct{}, 1),
 		mainThreadSaturation:  newSaturationMetric([]string{"raft", "thread", "main", "saturation"}, 1*time.Second),
 		preVoteDisabled:       conf.PreVoteDisabled || !transportSupportPreVote,
+		fastRecovery:          conf.FastRecovery,
 	}
 	if !transportSupportPreVote && !conf.PreVoteDisabled {
 		r.logger.Warn("pre-vote is disabled because it is not supported by the Transport")
@@ -606,6 +611,8 @@ func NewRaft(conf *Config, fsm FSM, logs LogStore, stable StableStore, snaps Sna
 	// to be called concurrently with a blocking RPC.
 	trans.SetHeartbeatHandler(r.processHeartbeat)
 
+	r.recoverFromCommitedLogs()
+
 	if conf.skipStartup {
 		return r, nil
 	}
@@ -697,6 +704,29 @@ func (r *Raft) tryRestoreSingleSnapshot(snapshot *SnapshotMeta) bool {
 	return true
 }
 
+// recoverFromCommitedLogs recovers the Raft node from committed logs.
+func (r *Raft) recoverFromCommitedLogs() error {
+	if !r.fastRecovery {
+		return nil
+	}
+	// If the store implements CommitTrackingLogStore, we can read the commit index from the store.
+	// This is useful when the store is able to track the commit index and we can avoid replaying logs.
+	store, ok := r.logs.(CommitTrackingLogStore)
+	if !ok {
+		return nil
+	}
+	commitIndex, err := store.GetCommitIndex()
+	if err != nil {
+		return fmt.Errorf("failed to read commit index from store: %w", err)
+	}
+	if commitIndex == 0 {
+		return nil
+	}
+
+	r.processLogs(commitIndex, nil)
+	return nil
+}
+
 func (r *Raft) config() Config {
 	return r.conf.Load().(Config)
 }

config.go

@@ -235,6 +235,11 @@ type Config struct {
 	// PreVoteDisabled deactivate the pre-vote feature when set to true
 	PreVoteDisabled bool
 
+	// FastRecovery controls if the Raft server should use the fast recovery
+	// mechanism. This mechanism allows a server to apply logs to the FSM till
+	// the last committed log
+	FastRecovery bool
+
 	// skipStartup allows NewRaft() to bypass all background work goroutines
 	skipStartup bool
 }

inmem_store.go

@@ -131,3 +131,31 @@ func (i *InmemStore) GetUint64(key []byte) (uint64, error) {
 	defer i.l.RUnlock()
 	return i.kvInt[string(key)], nil
 }
+
+type InmemCommitTrackingStore struct {
+	*InmemStore
+	commitIndexLock sync.RWMutex
+	commitIndex     uint64
+}
+
+// NewInmemCommitTrackingStore returns a new in-memory backend that tracks the commit index. Do not ever
+// use for production. Only for testing.
+func NewInmemCommitTrackingStore() *InmemCommitTrackingStore {
+	i := &InmemCommitTrackingStore{
+		InmemStore: NewInmemStore(),
+	}
+	return i
+}
+
+func (i *InmemCommitTrackingStore) SetCommitIndex(index uint64) error {
+	i.commitIndexLock.Lock()
+	defer i.commitIndexLock.Unlock()
+	i.commitIndex = index
+	return nil
+}
+
+func (i *InmemCommitTrackingStore) GetCommitIndex() (uint64, error) {
+	i.commitIndexLock.RLock()
+	defer i.commitIndexLock.RUnlock()
+	return i.commitIndex, nil
+}

log.go

@@ -190,3 +190,8 @@ func emitLogStoreMetrics(s LogStore, prefix []string, interval time.Duration, st
 		}
 	}
 }
+
+type CommitTrackingLogStore interface {
+	SetCommitIndex(idx uint64) error
+	GetCommitIndex() (uint64, error)
+}

raft.go

@@ -1354,6 +1354,8 @@ func (r *Raft) processLogs(index uint64, futures map[uint64]*logFuture) {
 		applyBatch(batch)
 	}
 
+	r.persistCommitIndex(index)
+
 	// Update the lastApplied index and term
 	r.setLastApplied(index)
 }
@@ -1385,6 +1387,20 @@ func (r *Raft) prepareLog(l *Log, future *logFuture) *commitTuple {
 	return nil
 }
 
+// persistCommitIndex updates the commit index in persist store if fast recovery is enabled.
+func (r *Raft) persistCommitIndex(index uint64) {
+	if !r.fastRecovery {
+		return
+	}
+	store, ok := r.logs.(CommitTrackingLogStore)
+	if !ok {
+		return
+	}
+	if err := store.SetCommitIndex(index); err != nil {
+		r.logger.Error("failed to set commit index in commit tracking log store", "index", index, "error", err)
+	}
+}
+
 // processRPC is called to handle an incoming RPC request. This must only be
 // called from the main thread.
 func (r *Raft) processRPC(rpc RPC) {