Enhancement: persist commit index in LogStore to accelerate recovery

api.go

@@ -73,6 +73,10 @@ var (
 	// ErrLeadershipTransferInProgress is returned when the leader is rejecting
 	// client requests because it is attempting to transfer leadership.
 	ErrLeadershipTransferInProgress = errors.New("leadership transfer in progress")
+
+	// ErrIncompatibleLogStore is returned when the log store does not support
+	// or implement some required methods.
+	ErrIncompatibleLogStore = errors.New("log store does not implement some required methods or malformed")
 )
 
 // Raft implements a Raft node.
@@ -222,6 +226,10 @@ type Raft struct {
 	// legacy metrics are those that have `_peer_name` as metric suffix instead as labels.
 	// e.g: raft_replication_heartbeat_peer0
 	noLegacyTelemetry bool
+
+	// RestoreCommittedLogs is used to enable restore committed logs mode
+	// restore committed logs mode is disabled if set to false.
+	RestoreCommittedLogs bool
 }
 
 // BootstrapCluster initializes a server's storage with the given cluster
@@ -575,6 +583,7 @@ func NewRaft(conf *Config, fsm FSM, logs LogStore, stable StableStore, snaps Sna
 		mainThreadSaturation:  newSaturationMetric([]string{"raft", "thread", "main", "saturation"}, 1*time.Second),
 		preVoteDisabled:       conf.PreVoteDisabled || !transportSupportPreVote,
 		noLegacyTelemetry:     conf.NoLegacyTelemetry,
+		RestoreCommittedLogs:  conf.RestoreCommittedLogs,
 	}
 	if !transportSupportPreVote && !conf.PreVoteDisabled {
 		r.logger.Warn("pre-vote is disabled because it is not supported by the Transport")
@@ -594,9 +603,14 @@ func NewRaft(conf *Config, fsm FSM, logs LogStore, stable StableStore, snaps Sna
 		return nil, err
 	}
 
+	if err := r.restoreFromCommittedLogs(); err != nil {
+		return nil, err
+	}
+
 	// Scan through the log for any configuration change entries.
 	snapshotIndex, _ := r.getLastSnapshot()
-	for index := snapshotIndex + 1; index <= lastLog.Index; index++ {
+	lastappliedIndex := r.getLastApplied()
+	for index := max(snapshotIndex, lastappliedIndex) + 1; index <= lastLog.Index; index++ {
 		var entry Log
 		if err := r.logs.GetLog(index, &entry); err != nil {
 			r.logger.Error("failed to get log", "index", index, "error", err)
@@ -706,6 +720,40 @@ func (r *Raft) tryRestoreSingleSnapshot(snapshot *SnapshotMeta) bool {
 	return true
 }
 
+// restoreFromCommittedLogs recovers the Raft node from committed logs.
+func (r *Raft) restoreFromCommittedLogs() error {
+	if !r.RestoreCommittedLogs {
+		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 {
+		r.logger.Warn("restore committed logs enabled but log store does not support it", "log_store", fmt.Sprintf("%T", r.logs))
+		return ErrIncompatibleLogStore
+	}
+
+	commitIndex, err := store.GetCommitIndex()
+	if err != nil {
+		r.logger.Error("failed to get commit index from store", "error", err)
+		return err
+	}
+
+	lastIndex, err := r.logs.LastIndex()
+	if err != nil {
+		r.logger.Error("failed to get last log index from store", "error", err)
+		return err
+	}
+	if commitIndex > lastIndex {
+		commitIndex = lastIndex
+	}
+
+	r.setCommitIndex(commitIndex)
+	r.processLogs(commitIndex, nil)
+	return nil
+}
+
 func (r *Raft) config() Config {
 	return r.conf.Load().(Config)
 }

config.go

@@ -240,6 +240,19 @@ type Config struct {
 	// e.g: raft_replication_heartbeat_peer0
 	NoLegacyTelemetry bool
 
+	// RestoreCommittedLogs controls if the Raft server should use the restore committed logs
+	// mechanism. Restore committed logs requires a LogStore implementation that
+	// support commit tracking. When such a store is used and this config
+	// enabled, raft nodes will replay all known-committed logs on disk
+	// before completing `NewRaft` on startup. This is mainly useful where
+	// the application allows relaxed-consistency reads from followers as it
+	// will reduce how far behind the follower's FSM is when it starts. If all reads
+	// are forwarded to the leader then there won't be observable benefit from this feature.
+	//
+	// Notice: If this is enabled, the log store MUST implement the CommitTrackingLogStore
+	// interface. Otherwise, Raft will fail to start and return ErrIncompatibleLogStore.
+	RestoreCommittedLogs bool
+
 	// skipStartup allows NewRaft() to bypass all background work goroutines
 	skipStartup bool
 }

inmem_store.go

@@ -6,8 +6,11 @@ package raft
 import (
 	"errors"
 	"sync"
+	"sync/atomic"
 )
 
+var _ CommitTrackingLogStore = &InmemCommitTrackingStore{}
+
 // InmemStore implements the LogStore and StableStore interface.
 // It should NOT EVER be used for production. It is used only for
 // unit tests. Use the MDBStore implementation instead.
@@ -131,3 +134,30 @@ func (i *InmemStore) GetUint64(key []byte) (uint64, error) {
 	defer i.l.RUnlock()
 	return i.kvInt[string(key)], nil
 }
+
+type commitIndexTrackingLog struct {
+	log         *Log
+	CommitIndex uint64
+}
+type InmemCommitTrackingStore struct {
+	InmemStore
+	commitIndex atomic.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) StageCommitIndex(index uint64) error {
+	i.commitIndex.Store(index)
+	return nil
+}
+
+func (i *InmemCommitTrackingStore) GetCommitIndex() (uint64, error) {
+	return i.commitIndex.Load(), nil
+}

log.go

@@ -190,3 +190,32 @@ func emitLogStoreMetrics(s LogStore, prefix []string, interval time.Duration, st
 		}
 	}
 }
+
+type CommitTrackingLogStore interface {
+	LogStore
+
+	// StageCommitIndex stages a new commit index to be persisted.
+	// The staged commit index MUST only be persisted in a manner that is atomic
+	// with the following StoreLogs call in the face of a crash.
+	// This allows the Raft implementation to optimize commit index updates
+	// without risking inconsistency between the commit index and the log entries.
+	//
+	// The implementation MUST NOT persist this value separately from the log entries.
+	// Instead, it should stage the value to be written atomically with the next
+	// StoreLogs call.
+	//
+	// GetCommitIndex MUST never return a value higher than the last index in the log,
+	// even if a higher value has been staged with this method.
+	//
+	// idx is the new commit index to stage.
+	StageCommitIndex(idx uint64) error
+
+	// GetCommitIndex returns the latest persisted commit index from the latest log entry
+	// in the store at startup.
+	//
+	// GetCommitIndex should not return a value higher than the last index in the log.
+	// If that happens, the last index in the log will be used.
+	//
+	// When no commit index is found in the log store, return (0, nil).
+	GetCommitIndex() (uint64, error)
+}

raft.go

@@ -1262,6 +1262,9 @@ func (r *Raft) dispatchLogs(applyLogs []*logFuture) {
 		r.leaderState.inflight.PushBack(applyLog)
 	}
 
+	commitIndex := r.getCommitIndex()
+	r.tryStageCommitIndex(commitIndex)
+
 	// Write the log entry locally
 	if err := r.logs.StoreLogs(logs); err != nil {
 		r.logger.Error("failed to commit logs", "error", err)
@@ -1385,6 +1388,20 @@ func (r *Raft) prepareLog(l *Log, future *logFuture) *commitTuple {
 	return nil
 }
 
+// tryStageCommitIndex updates the commit index in persist store if restore committed logs is enabled and log store implements CommitTrackingLogStore.
+func (r *Raft) tryStageCommitIndex(commitIndex uint64) {
+	if !r.RestoreCommittedLogs {
+		return
+	}
+	store, ok := r.logs.(CommitTrackingLogStore)
+	if !ok {
+		return
+	}
+	if err := store.StageCommitIndex(commitIndex); err != nil {
+		r.logger.Error("failed to stage commit index in commit tracking log store", "index", commitIndex, "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) {
@@ -1535,6 +1552,11 @@ func (r *Raft) appendEntries(rpc RPC, a *AppendEntriesRequest) {
 		}
 
 		if n := len(newEntries); n > 0 {
+			// Stage the future commit index if possible
+			lastNewIndex := newEntries[len(newEntries)-1].Index
+			commitIndex := min(a.LeaderCommitIndex, lastNewIndex)
+			r.tryStageCommitIndex(commitIndex)
+
 			// Append the new entries
 			if err := r.logs.StoreLogs(newEntries); err != nil {
 				r.logger.Error("failed to append to logs", "error", err)