fix: keep lastLog cache in sync after append truncation failures

raft.go

@@ -1539,8 +1539,7 @@ func (r *Raft) appendEntries(rpc RPC, a *AppendEntriesRequest) {
 			// Append the new entries
 			if err := r.logs.StoreLogs(newEntries); err != nil {
 				r.logger.Error("failed to append to logs", "error", err)
-				// TODO: leaving r.getLastLog() in the wrong
-				// state if there was a truncation above
+				r.refreshLastLogFromStore()
 				return
 			}
 
@@ -1580,6 +1579,28 @@ func (r *Raft) appendEntries(rpc RPC, a *AppendEntriesRequest) {
 	r.setLastContact()
 }
 
+func (r *Raft) refreshLastLogFromStore() {
+	lastIdx, lastTerm := r.getLastSnapshot()
+
+	storeLastIdx, err := r.logs.LastIndex()
+	if err != nil {
+		r.logger.Error("failed to read last index after append failure", "error", err)
+		r.setLastLog(lastIdx, lastTerm)
+		return
+	}
+	if storeLastIdx > 0 {
+		var last Log
+		if err := r.logs.GetLog(storeLastIdx, &last); err != nil {
+			r.logger.Error("failed to read last log after append failure", "index", storeLastIdx, "error", err)
+		} else {
+			lastIdx = last.Index
+			lastTerm = last.Term
+		}
+	}
+
+	r.setLastLog(lastIdx, lastTerm)
+}
+
 // processConfigurationLogEntry takes a log entry and updates the latest
 // configuration if the entry results in a new configuration. This must only be
 // called from the main thread, or from NewRaft() before any threads have begun.

raft_test.go

@@ -2002,6 +2002,18 @@ func TestRaft_NotifyCh(t *testing.T) {
 	}
 }
 
+type appendFailingLogStore struct {
+	*InmemStore
+	failStoreLogs bool
+}
+
+func (s *appendFailingLogStore) StoreLogs(logs []*Log) error {
+	if s.failStoreLogs {
+		return errors.New("injected StoreLogs failure")
+	}
+	return s.InmemStore.StoreLogs(logs)
+}
+
 func TestRaft_AppendEntry(t *testing.T) {
 	c := MakeCluster(3, t, nil)
 	defer c.Close()
@@ -2061,6 +2073,68 @@ func TestRaft_AppendEntry(t *testing.T) {
 	require.True(t, resp2.Success)
 }
 
+func TestRaft_AppendEntriesStoreLogsFailureRefreshesLastLogAfterTruncate(t *testing.T) {
+	_, transport := NewInmemTransport("")
+	logs := &appendFailingLogStore{InmemStore: NewInmemStore()}
+	require.NoError(t, logs.StoreLogs([]*Log{
+		{Index: 1, Term: 1, Type: LogCommand, Data: []byte("1")},
+		{Index: 2, Term: 1, Type: LogCommand, Data: []byte("2")},
+		{Index: 3, Term: 1, Type: LogCommand, Data: []byte("3")},
+		{Index: 4, Term: 1, Type: LogCommand, Data: []byte("4")},
+		{Index: 5, Term: 1, Type: LogCommand, Data: []byte("5")},
+	}))
+	logs.failStoreLogs = true
+
+	r := &Raft{
+		trans:     transport,
+		logs:      logs,
+		logger:    hclog.New(nil),
+		localID:   "local",
+		localAddr: transport.LocalAddr(),
+	}
+	cfg := *DefaultConfig()
+	cfg.LocalID = r.localID
+	r.conf.Store(cfg)
+	r.raftState.setCurrentTerm(5)
+	r.setState(Follower)
+	r.setLastLog(5, 1)
+
+	leaderID := ServerID("leader-id")
+	leaderAddr := ServerAddress("leader-addr")
+	encodedLeader := transport.EncodePeer(leaderID, leaderAddr)
+	req := &AppendEntriesRequest{
+		RPCHeader: RPCHeader{
+			ProtocolVersion: cfg.ProtocolVersion,
+			ID:              []byte(leaderID),
+			Addr:            encodedLeader,
+		},
+		Term:         5,
+		Leader:       encodedLeader,
+		PrevLogEntry: 2,
+		PrevLogTerm:  1,
+		Entries: []*Log{
+			{Index: 3, Term: 2, Type: LogCommand, Data: []byte("replacement-3")},
+		},
+	}
+
+	chResp := make(chan RPCResponse, 1)
+	r.appendEntries(RPC{RespChan: chResp}, req)
+	resp := <-chResp
+	require.NoError(t, resp.Error)
+
+	appendResp, ok := resp.Response.(*AppendEntriesResponse)
+	require.True(t, ok)
+	require.False(t, appendResp.Success)
+
+	lastIdx, lastTerm := r.getLastLog()
+	require.Equal(t, uint64(2), lastIdx, "last log index should match durable log store after truncation+failure")
+	require.Equal(t, uint64(1), lastTerm)
+
+	storeLastIdx, err := logs.LastIndex()
+	require.NoError(t, err)
+	require.Equal(t, uint64(2), storeLastIdx)
+}
+
 // TestRaft_PreVoteMixedCluster focus on testing a cluster with
 // a mix of nodes that have pre-vote activated and deactivated.
 // Once the cluster is created, we force an election by partioning the leader