diff --git a/raft.go b/raft.go index c95f9e93..25e40264 100644 --- a/raft.go +++ b/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. diff --git a/raft_test.go b/raft_test.go index 9cfbf6c2..2c9744d9 100644 --- a/raft_test.go +++ b/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