bft: restore the start view from the trailing proposal record

internal/bft/state.go

@@ -8,6 +8,7 @@ package bft
 import (
 	"errors"
 	"fmt"
+	"slices"
 	"time"
 
 	"github.com/hyperledger/SmartBFT/pkg/api"
@@ -78,15 +79,62 @@ func (ps *PersistedState) LoadNewViewIfApplicable() (*types.ViewAndSeq, error) {
 		ps.Logger.Infof("Nothing to restore")
 		return nil, nil
 	}
-	lastEntry := entries[len(entries)-1]
-	lastPersistedMessage := &protos.SavedMessage{}
-	if err := proto.Unmarshal(lastEntry, lastPersistedMessage); err != nil {
-		ps.Logger.Errorf("Failed unmarshaling last entry from WAL: %v", err)
-		return nil, fmt.Errorf("failed unmarshaling last entry from WAL: %w", err)
+
+	// A proposal or a commit can be appended after the new view record, so the
+	// new view is not necessarily the last entry. Walk backwards to the latest
+	// view transition.
+	for i := range slices.Backward(entries) {
+		persistedMessage := &protos.SavedMessage{}
+		if err := proto.Unmarshal(entries[i], persistedMessage); err != nil {
+			ps.Logger.Errorf("Failed unmarshaling entry %d from WAL: %v", i, err)
+			return nil, fmt.Errorf("failed unmarshaling entry %d from WAL: %w", i, err)
+		}
+		if newViewMsg := persistedMessage.GetNewView(); newViewMsg != nil {
+			ps.Logger.Infof("found a newView record in WAL")
+			return &types.ViewAndSeq{View: newViewMsg.GetViewId(), Seq: newViewMsg.GetLatestSequence()}, nil
+		}
+		if persistedMessage.GetViewChange() != nil {
+			// A later view change supersedes any new view before it.
+			// LoadViewChangeIfApplicable sets the starting view in that case.
+			ps.Logger.Infof("found a viewChange record after the last newView record")
+			return nil, nil
+		}
 	}
-	if newViewMsg := lastPersistedMessage.GetNewView(); newViewMsg != nil {
-		ps.Logger.Infof("last entry in WAL is a newView record")
-		return &types.ViewAndSeq{View: newViewMsg.GetViewId(), Seq: newViewMsg.GetLatestSequence()}, nil
+	return nil, nil
+}
+
+// LoadLastProposalView returns the metadata of the latest proposal record.
+// The built-in WAL drops a NewView record when the next pre-prepare is saved
+// with truncateTo=true, so after a restart the proposal can be the only record
+// that defines the view. A later NewView or ViewChange record takes precedence
+// and makes this method return nil: LoadNewViewIfApplicable and
+// LoadViewChangeIfApplicable handle those records.
+func (ps *PersistedState) LoadLastProposalView() (*protos.ViewMetadata, error) {
+	for i := range slices.Backward(ps.Entries) {
+		persistedMessage := &protos.SavedMessage{}
+		if err := proto.Unmarshal(ps.Entries[i], persistedMessage); err != nil {
+			ps.Logger.Errorf("Failed unmarshaling entry %d from WAL: %v", i, err)
+			return nil, fmt.Errorf("failed unmarshaling entry %d from WAL: %w", i, err)
+		}
+		if persistedMessage.GetNewView() != nil || persistedMessage.GetViewChange() != nil {
+			return nil, nil
+		}
+		proposed := persistedMessage.GetProposedRecord()
+		if proposed == nil {
+			// A commit record always follows its proposal record, so keep
+			// walking backwards to find the proposal that carries the metadata.
+			continue
+		}
+		prePrepare := proposed.GetPrePrepare()
+		if prePrepare == nil {
+			return nil, errors.New("proposed record has no pre-prepare")
+		}
+
+		metadata := &protos.ViewMetadata{}
+		if err := proto.Unmarshal(prePrepare.GetProposal().GetMetadata(), metadata); err != nil {
+			return nil, fmt.Errorf("failed unmarshaling proposal metadata: %w", err)
+		}
+		return metadata, nil
 	}
 	return nil, nil
 }

internal/bft/state_test.go

@@ -314,6 +314,8 @@ func TestStateLoadIfApplicable(t *testing.T) {
 		},
 	}
 
+	commitRecord := newCommitRecord(11, 5)
+
 	for _, testCase := range []struct {
 		description        string
 		WALContent         [][]byte
@@ -327,11 +329,25 @@ func TestStateLoadIfApplicable(t *testing.T) {
 		{
 			description:   "malformed record",
 			WALContent:    [][]byte{{1, 2, 3}},
-			expectedError: "failed unmarshaling last entry from WAL: proto:",
+			expectedError: "failed unmarshaling",
 		},
 		{
-			description: "last record is a proposal",
+			description: "new view followed by a proposal",
 			WALContent:  [][]byte{bft.MarshalOrPanic(newView), bft.MarshalOrPanic(proposedRecord)},
+			// The proposal belongs to the new view, so the new view still
+			// defines the starting view.
+			expectedViewAndSeq: &types.ViewAndSeq{View: 5, Seq: 10},
+		},
+		{
+			description:        "new view followed by a commit",
+			WALContent:         [][]byte{bft.MarshalOrPanic(newView), bft.MarshalOrPanic(commitRecord)},
+			expectedViewAndSeq: &types.ViewAndSeq{View: 5, Seq: 10},
+		},
+		{
+			description:        "new view followed by a proposal and a view change",
+			WALContent:         [][]byte{bft.MarshalOrPanic(newView), bft.MarshalOrPanic(proposedRecord), bft.MarshalOrPanic(savedViewChange)},
+			expectedViewChange: viewChange,
+			// The later view change supersedes the new view.
 		},
 		{
 			description:        "last record is a new view",
@@ -376,6 +392,137 @@ func TestStateLoadIfApplicable(t *testing.T) {
 	}
 }
 
+func TestLoadLastProposalView(t *testing.T) {
+	proposalMetadata := &protos.ViewMetadata{ViewId: 5, LatestSequence: 10, DecisionsInView: 3}
+	proposal := newProposalRecord(11, 5, proposalMetadata)
+	commit := newCommitRecord(11, 5)
+	newView := newViewRecord(6, 10)
+	viewChange := &protos.SavedMessage{
+		Content: &protos.SavedMessage_ViewChange{
+			ViewChange: &protos.ViewChange{NextView: 6},
+		},
+	}
+
+	for _, testCase := range []struct {
+		description      string
+		WALContent       [][]byte
+		expectedMetadata *protos.ViewMetadata
+		expectedError    string
+	}{
+		{
+			description: "empty",
+		},
+		{
+			description:      "last record is a proposal",
+			WALContent:       [][]byte{bft.MarshalOrPanic(proposal)},
+			expectedMetadata: proposalMetadata,
+		},
+		{
+			description:      "commit after its proposal",
+			WALContent:       [][]byte{bft.MarshalOrPanic(proposal), bft.MarshalOrPanic(commit)},
+			expectedMetadata: proposalMetadata,
+		},
+		{
+			description: "commit without a proposal",
+			WALContent:  [][]byte{bft.MarshalOrPanic(commit)},
+		},
+		{
+			description:      "new view followed by a proposal",
+			WALContent:       [][]byte{bft.MarshalOrPanic(newView), bft.MarshalOrPanic(proposal)},
+			expectedMetadata: proposalMetadata,
+		},
+		{
+			description: "proposal followed by a view change",
+			WALContent:  [][]byte{bft.MarshalOrPanic(proposal), bft.MarshalOrPanic(viewChange)},
+		},
+		{
+			description: "commit followed by a view change",
+			WALContent:  [][]byte{bft.MarshalOrPanic(proposal), bft.MarshalOrPanic(commit), bft.MarshalOrPanic(viewChange)},
+		},
+		{
+			description: "proposal followed by a new view",
+			WALContent:  [][]byte{bft.MarshalOrPanic(proposal), bft.MarshalOrPanic(newView)},
+		},
+		{
+			description:   "malformed record",
+			WALContent:    [][]byte{{1, 2, 3}},
+			expectedError: "failed unmarshaling",
+		},
+		{
+			description: "proposal without a pre-prepare",
+			WALContent: [][]byte{bft.MarshalOrPanic(&protos.SavedMessage{
+				Content: &protos.SavedMessage_ProposedRecord{
+					ProposedRecord: &protos.ProposedRecord{},
+				},
+			})},
+			expectedError: "proposed record has no pre-prepare",
+		},
+	} {
+		t.Run(testCase.description, func(t *testing.T) {
+			state := &bft.PersistedState{
+				Entries: testCase.WALContent,
+				Logger:  zap.NewNop().Sugar(),
+			}
+
+			metadata, err := state.LoadLastProposalView()
+			if testCase.expectedError != "" {
+				require.ErrorContains(t, err, testCase.expectedError)
+				return
+			}
+
+			require.NoError(t, err)
+			if testCase.expectedMetadata == nil {
+				require.Nil(t, metadata)
+				return
+			}
+			require.NotNil(t, metadata)
+			require.Equal(t, testCase.expectedMetadata.GetViewId(), metadata.GetViewId())
+			require.Equal(t, testCase.expectedMetadata.GetDecisionsInView(), metadata.GetDecisionsInView())
+		})
+	}
+}
+
+// When a proposal follows a NewView record, the new view record still defines
+// the starting view. Restore uses the proposal record to restore the in-flight
+// state and may change View.Number, so both must agree.
+func TestRestoreStartViewMatchesNewViewRecord(t *testing.T) {
+	basicLog, err := zap.NewDevelopment()
+	require.NoError(t, err)
+
+	entries := [][]byte{
+		bft.MarshalOrPanic(newViewRecord(301, 200)),
+		bft.MarshalOrPanic(newProposalRecord(201, 301, &protos.ViewMetadata{
+			ViewId:         301,
+			LatestSequence: 200,
+		})),
+	}
+
+	state := &bft.PersistedState{
+		Entries:          entries,
+		Logger:           basicLog.Sugar(),
+		InFlightProposal: &bft.InFlightData{},
+	}
+
+	viewAndSeq, err := state.LoadNewViewIfApplicable()
+	require.NoError(t, err)
+	require.NotNil(t, viewAndSeq)
+	require.Equal(t, uint64(301), viewAndSeq.View)
+	require.Equal(t, uint64(200), viewAndSeq.Seq)
+
+	verifier := &mocks.VerifierMock{}
+	verifier.On("RequestsFromProposal", mock.Anything).Return(nil)
+	view := &bft.View{
+		Number:           300,
+		ProposalSequence: 201,
+		MetricsView:      api.NewMetricsView(&disabled.Provider{}),
+		Verifier:         verifier,
+	}
+
+	require.NoError(t, state.Restore(view))
+	require.Equal(t, uint64(301), view.Number)
+	require.Equal(t, viewAndSeq.View, view.Number)
+}
+
 // recordingWAL stands in for the write ahead log at the api.WriteAheadLog seam
 // and keeps, in order, every appended record together with its truncation flag.
 type recordingWAL struct {

pkg/consensus/consensus.go

@@ -507,6 +507,19 @@ func (c *Consensus) setViewAndSeq(view, seq, dec uint64) (newView, newSeq, newDe
 	}
 	if viewSeq == nil {
 		c.Logger.Debugf("No new view to restore")
+
+		// The built-in WAL truncates a NewView record when the next pre-prepare
+		// is saved. In that case the trailing proposal still defines the view
+		// and the decision count the node was in before the restart.
+		lastProposal, err := c.state.LoadLastProposalView()
+		if err != nil {
+			c.Logger.Panicf("Failed loading the last proposal view, error: %v", err)
+		}
+		if lastProposal != nil && lastProposal.GetViewId() > newView {
+			c.Logger.Debugf("Restoring start view %d from the last proposal record", lastProposal.GetViewId())
+			newView = lastProposal.GetViewId()
+			newDec = lastProposal.GetDecisionsInView()
+		}
 	} else if viewSeq.Seq >= seq {
 		// Check if metadata should be taken from the restored new view
 		c.Logger.Debugf("Restoring from new view with view %d and seq %d", viewSeq.View, viewSeq.Seq)

pkg/consensus/consensus_test.go

@@ -10,6 +10,9 @@ import (
 	"time"
 
 	algorithm "github.com/hyperledger/SmartBFT/internal/bft"
+	"github.com/hyperledger/SmartBFT/smartbftprotos"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
 )
 
 // TestComplainDoesNotBlockDuringReconfiguration covers the shutdown deadlock:
@@ -33,3 +36,65 @@ func TestComplainDoesNotBlockDuringReconfiguration(t *testing.T) {
 		t.Fatal("Complain blocked while a reconfiguration held the lock")
 	}
 }
+
+func TestSetViewAndSeqUsesTheTrailingProposalView(t *testing.T) {
+	logger := zap.NewNop().Sugar()
+
+	for _, testCase := range []struct {
+		description       string
+		proposalView      uint64
+		proposalDecisions uint64
+		expectedView      uint64
+		expectedDecisions uint64
+	}{
+		{
+			description:       "newer view in the proposal record",
+			proposalView:      1,
+			proposalDecisions: 5,
+			expectedView:      1,
+			expectedDecisions: 5,
+		},
+		{
+			description:       "same view as the application metadata",
+			proposalView:      0,
+			proposalDecisions: 5,
+			expectedView:      0,
+			expectedDecisions: 3,
+		},
+	} {
+		t.Run(testCase.description, func(t *testing.T) {
+			proposal := &smartbftprotos.SavedMessage{
+				Content: &smartbftprotos.SavedMessage_ProposedRecord{
+					ProposedRecord: &smartbftprotos.ProposedRecord{
+						PrePrepare: &smartbftprotos.PrePrepare{
+							Seq:  11,
+							View: testCase.proposalView,
+							Proposal: &smartbftprotos.Proposal{
+								Metadata: algorithm.MarshalOrPanic(&smartbftprotos.ViewMetadata{
+									ViewId:          testCase.proposalView,
+									LatestSequence:  10,
+									DecisionsInView: testCase.proposalDecisions,
+								}),
+							},
+						},
+					},
+				},
+			}
+
+			consensus := &Consensus{
+				Logger: logger,
+				state: &algorithm.PersistedState{
+					Entries: [][]byte{algorithm.MarshalOrPanic(proposal)},
+					Logger:  logger,
+				},
+			}
+
+			// The application metadata is at sequence 10 with 2 decisions, so
+			// setViewAndSeq starts with 3 decisions and the view from metadata.
+			view, seq, dec := consensus.setViewAndSeq(0, 10, 2)
+			require.Equal(t, testCase.expectedView, view)
+			require.Equal(t, uint64(10), seq)
+			require.Equal(t, testCase.expectedDecisions, dec)
+		})
+	}
+}