diff --git a/internal/bft/state.go b/internal/bft/state.go index 366ca309..c2edd841 100644 --- a/internal/bft/state.go +++ b/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 } diff --git a/internal/bft/state_test.go b/internal/bft/state_test.go index 16c1e279..6c4f41c1 100644 --- a/internal/bft/state_test.go +++ b/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 { diff --git a/pkg/consensus/consensus.go b/pkg/consensus/consensus.go index 058ca5a2..267deb84 100644 --- a/pkg/consensus/consensus.go +++ b/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) diff --git a/pkg/consensus/consensus_test.go b/pkg/consensus/consensus_test.go index 9e527704..3d2f4149 100644 --- a/pkg/consensus/consensus_test.go +++ b/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) + }) + } +} diff --git a/test/start_view_recovery_test.go b/test/start_view_recovery_test.go new file mode 100644 index 00000000..3e39bbfc --- /dev/null +++ b/test/start_view_recovery_test.go @@ -0,0 +1,90 @@ +// Copyright IBM Corp. All Rights Reserved. +// +// SPDX-License-Identifier: Apache-2.0 +// + +package test + +import ( + "strings" + "testing" + "time" + + "github.com/hyperledger/SmartBFT/smartbftprotos" + "github.com/stretchr/testify/require" + "go.uber.org/zap" + "go.uber.org/zap/zapcore" +) + +func TestRestartAfterNewViewAndProposalRestoresStartView(t *testing.T) { + t.Parallel() + network := NewNetwork() + defer network.Shutdown() + + testDir := t.TempDir() + + const numberOfNodes = 4 + nodes := make([]*App, 0, numberOfNodes) + + for i := 1; i <= numberOfNodes; i++ { + n := newNode(uint64(i), network, t.Name(), testDir, false, 0) + if i == numberOfNodes { + // Keep the node at PROPOSED: it accepts the pre-prepare and sends + // its prepare, but cannot collect a quorum of prepares or commits. + n.LoseMessages(func(msg *smartbftprotos.Message) bool { + return msg.GetPrepare() != nil || msg.GetCommit() != nil + }) + } else { + // A non-nil filter suppresses the test app's check for prepares + // when it signs an in-flight proposal. + n.LoseMessages(func(*smartbftprotos.Message) bool { return false }) + } + nodes = append(nodes, n) + } + + restarted := nodes[numberOfNodes-1] + proposalProcessed := make(chan struct{}, 1) + + baseLogger := restarted.Consensus.Logger.(*zap.SugaredLogger).Desugar() + restarted.Consensus.Logger = baseLogger.WithOptions(zap.Hooks(func(entry zapcore.Entry) error { + if strings.Contains(entry.Message, "Processed proposal with seq") { + select { + case proposalProcessed <- struct{}{}: + default: + } + } + return nil + })).Sugar() + + startNodes(nodes, network) + + // Move every node to view 1. The restarted node writes a ViewChange and a + // NewView record. + for _, n := range nodes { + n.Consensus.Complain(0, true) + } + + require.Eventually(t, func() bool { + return nodes[0].Consensus.GetLeaderID() == 2 + }, 10*time.Second, 10*time.Millisecond, "the cluster did not start view 1") + + // The leader of view 1 proposes. The restarted node processes the + // pre-prepare. Saving the pre-prepare truncates the earlier ViewChange and + // NewView records, so its WAL ends with ProposedRecord. + nodes[1].Submit(Request{ID: "view 1 proposal", ClientID: "alice"}) + + select { + case <-proposalProcessed: + case <-time.After(10 * time.Second): + t.Fatal("the restarted node did not process the proposal in view 1") + } + + restarted.RestartSync(false) + + require.Eventually(t, func() bool { + return restarted.Consensus.GetLeaderID() != 0 + }, 10*time.Second, 10*time.Millisecond, "the restarted node did not start") + + require.Equal(t, uint64(2), restarted.Consensus.GetLeaderID(), + "the restarted node started the controller in the view from the metadata, not the view from the WAL") +}