From a57afdda5e67312db8539f778f6f06ebc20466db Mon Sep 17 00:00:00 2001 From: Fedor Partanskiy Date: Tue, 3 Nov 2026 15:53:33 +0300 Subject: [PATCH] bft: restore the start view from the trailing proposal record A NewView record can be followed by a proposal or a commit when the node restarts before the proposal is decided. The built-in WAL saves the pre-prepare with truncateTo=true, which drops the NewView record. After a restart only the proposal pair remains. LoadNewViewIfApplicable returned nil, the controller started from the older application metadata, and Restore set View.Number to the proposal's view. The controller and the view disagreed about the current view and leader. Walk the WAL backwards to the latest view transition. Return the last NewView, and stop at a later ViewChange so it keeps precedence. When no NewView survives, take the view from the trailing ProposedRecord or Commit and start the controller there. Add unit tests for the fallback and an integration test that restarts a node after a NewView and a proposal, then checks that the controller starts in the restored view. Signed-off-by: Fedor Partanskiy --- internal/bft/state.go | 64 +++++++++++-- internal/bft/state_test.go | 151 ++++++++++++++++++++++++++++++- pkg/consensus/consensus.go | 13 +++ pkg/consensus/consensus_test.go | 65 +++++++++++++ test/start_view_recovery_test.go | 90 ++++++++++++++++++ 5 files changed, 373 insertions(+), 10 deletions(-) create mode 100644 test/start_view_recovery_test.go 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") +}