From e4f77c5850592aada930444d469c20ea5e3638ff Mon Sep 17 00:00:00 2001 From: Fedor Partanskiy Date: Mon, 12 Oct 2026 13:54:41 +0300 Subject: [PATCH] Truncate the WAL whenever no proposal is in flight The WAL only freed space at a truncation point, which only a pre-prepare carried. A node that never proposed never deleted a file: files rotated by size and truncateIndex stayed zero. PersistedState.Save now derives the truncation flag from whether a proposal is in flight rather than from the message type, so NewView after sync or view change and ViewChange with nothing in flight also truncate. A commit never does, keeping a pre-prepare and its commit together. Signed-off-by: Fedor Partanskiy --- internal/bft/controller.go | 241 +++++++-- internal/bft/controller_sync_test.go | 482 ++++++++++++++++++ internal/bft/controller_test.go | 97 +++- internal/bft/controller_view_guard_test.go | 156 ++++++ internal/bft/mocks/state.go | 22 +- internal/bft/mocks/view_controller.go | 25 +- internal/bft/state.go | 20 +- internal/bft/state_test.go | 298 +++++++++-- internal/bft/view.go | 16 +- internal/bft/view_test.go | 120 ++++- internal/bft/viewchanger.go | 92 +++- internal/bft/viewchanger_test.go | 547 +++++++++++++++++++-- test/test_app.go | 30 +- test/wal_truncation_test.go | 109 ++++ 14 files changed, 2113 insertions(+), 142 deletions(-) create mode 100644 internal/bft/controller_view_guard_test.go diff --git a/internal/bft/controller.go b/internal/bft/controller.go index 3bf110eb..6bab0f4a 100644 --- a/internal/bft/controller.go +++ b/internal/bft/controller.go @@ -108,12 +108,22 @@ type Controller struct { currViewLock sync.RWMutex currViewNumber uint64 + // minRestartView is the lowest view the controller may start. The view + // changer raises it when it leaves a view, so the sync path cannot restart + // that view behind the view changer's back. Zero means no view is held. + minRestartView uint64 + + // viewChangeLock serializes stopping or starting a view with writing a + // record that may truncate the WAL. A view changer holds it from + // AbortViewAndHold until ReleaseViewWrite, so no view can append a record + // while the view changer writes the record that must follow the abort. + viewChangeLock sync.Mutex + viewWriteHeld bool currDecisionsInViewLock sync.RWMutex currDecisionsInView uint64 - viewChange chan viewInfo - abortViewChan chan uint64 + viewChange chan viewInfo stopOnce sync.Once stopChan chan struct{} @@ -181,6 +191,32 @@ func (c *Controller) setCurrentViewNumber(viewNumber uint64) { c.currViewNumber = viewNumber } +// holdView keeps the controller from starting the given view again until the +// view changer asks for a new one. +func (c *Controller) holdView(view uint64) { + c.currViewLock.Lock() + defer c.currViewLock.Unlock() + + c.minRestartView = max(c.minRestartView, view+1) +} + +// releaseHeldView lets the controller start any view again. +func (c *Controller) releaseHeldView() { + c.currViewLock.Lock() + defer c.currViewLock.Unlock() + + c.minRestartView = 0 +} + +// isViewHeld tells whether the view changer has left the given view and the +// controller must not start it again. +func (c *Controller) isViewHeld(view uint64) bool { + c.currViewLock.RLock() + defer c.currViewLock.RUnlock() + + return view < c.minRestartView +} + func (c *Controller) getCurrentDecisionsInView() uint64 { c.currDecisionsInViewLock.RLock() defer c.currDecisionsInViewLock.RUnlock() @@ -384,12 +420,20 @@ func (c *Controller) startView(proposalSequence uint64) { } func (c *Controller) changeView(newViewNumber uint64, newProposalSequence uint64, newDecisionsInView uint64) { + c.viewChangeLock.Lock() + defer c.viewChangeLock.Unlock() + latestView := c.getCurrentViewNumber() if latestView > newViewNumber { c.Logger.Debugf("Got view change to %d but already at %d", newViewNumber, latestView) return } + if c.isViewHeld(newViewNumber) { + c.Logger.Debugf("Not starting view %d: the view changer has left it", newViewNumber) + return + } + leader := c.currentViewLeader() stopped := c.currentViewStopped() @@ -414,11 +458,20 @@ func (c *Controller) changeView(newViewNumber uint64, newProposalSequence uint64 } func (c *Controller) abortView(view uint64) bool { - currView := c.getCurrentViewNumber() - c.Logger.Debugf("view for abort %d, current view %d", view, currView) + currViewNum := c.getCurrentViewNumber() + c.Logger.Debugf("view for abort %d, current view %d", view, currViewNum) + + if view < currViewNum { + c.Logger.Debugf("Was asked to abort view %d but the current view with number %d", view, currViewNum) + return false + } + + c.currViewLock.RLock() + currView := c.currView + c.currViewLock.RUnlock() - if view < currView { - c.Logger.Debugf("Was asked to abort view %d but the current view with number %d", view, currView) + if currView == nil { + c.Logger.Debugf("Was asked to abort view %d but no view was started", view) return false } @@ -427,12 +480,19 @@ func (c *Controller) abortView(view uint64) bool { c.relinquishLeaderToken() // Kill current view - c.Logger.Debugf("Aborting current view with number %d", c.currViewNumber) - c.currView.Abort() + c.Logger.Debugf("Aborting current view with number %d", currViewNum) + currView.Abort() return true } +// abortCurrentView stops the current view and waits for it to finish. The +// caller must hold viewChangeLock, which keeps the stop and the following +// record write from being overtaken by a new view. +func (c *Controller) abortCurrentView() { + c.abortView(c.getCurrentViewNumber()) +} + // Sync initiates a synchronization func (c *Controller) Sync() { if iAmLeader, _ := c.iAmTheLeader(); iAmLeader { @@ -441,13 +501,79 @@ func (c *Controller) Sync() { c.grabSyncToken() } -// AbortView makes the controller abort the current view -func (c *Controller) AbortView(view uint64) { +// AbortView makes the controller abort the current view and waits until the +// view has stopped. Waiting is what makes it safe to persist state that must +// not be overtaken by a record written by the view, such as a truncating new +// view record. It returns false when the view is older than the current one or +// when the controller is stopping. It does not wait for the controller loop, +// which may be busy with a long sync or an application delivery. +func (c *Controller) AbortView(view uint64) bool { c.Logger.Debugf("AbortView, the current view num is %d", c.getCurrentViewNumber()) c.Batcher.Close() - c.abortViewChan <- view + c.viewChangeLock.Lock() + defer c.viewChangeLock.Unlock() + + if c.stopped() { + return false + } + + return c.abortView(view) +} + +// AbortViewAndHold aborts the current view, waits until it has stopped, and +// keeps the controller from starting that view again until the view changer +// asks for a new one. The sync path restarts views, so without the hold it +// could restart the view between the abort and the record the view changer +// writes next. +// +// On success the controller keeps viewChangeLock held, so no other view can be +// started or stopped until ReleaseViewWrite. The caller must call it after +// writing the record that had to follow the abort. +func (c *Controller) AbortViewAndHold(view uint64) bool { + c.Logger.Debugf("AbortViewAndHold, the current view num is %d", c.getCurrentViewNumber()) + + c.Batcher.Close() + + c.viewChangeLock.Lock() + + if c.stopped() { + c.viewChangeLock.Unlock() + + return false + } + + if !c.abortView(view) { + c.viewChangeLock.Unlock() + + return false + } + + c.holdView(view) + + c.currViewLock.Lock() + c.viewWriteHeld = true + c.currViewLock.Unlock() + + return true +} + +// ReleaseViewWrite lets the controller start or stop views again, once the view +// changer has written the record that had to follow AbortViewAndHold, and +// clears the hold that kept the left view stopped. +func (c *Controller) ReleaseViewWrite() { + c.currViewLock.Lock() + held := c.viewWriteHeld + c.viewWriteHeld = false + if held { + c.minRestartView = 0 + } + c.currViewLock.Unlock() + + if held { + c.viewChangeLock.Unlock() + } } // ViewChanged makes the controller abort the current view and start a new one with the given numbers @@ -488,9 +614,10 @@ func (c *Controller) run() { c.decide(d) case newView := <-c.viewChange: c.Logger.Debugf("get newView from viewChange") + // The view changer has finished leaving the old view and asks for + // this one, so it is safe to start it again. + c.releaseHeldView() c.changeView(newView.viewNumber, newView.proposalSeq, 0) - case view := <-c.abortViewChan: - c.abortView(view) case <-c.stopChan: return case <-c.leaderToken: @@ -657,30 +784,54 @@ func (c *Controller) sync() (viewNum uint64, seq uint64, decisions uint64) { // And the synchronizer did not return a new view return 0, 0, 0 } - } else { - if response.View <= controllerViewNum && latestDecisionViewNum < controllerViewNum { - return 0, 0, 0 // no new view to report + } else if response.View <= controllerViewNum && latestDecisionViewNum < controllerViewNum { + return 0, 0, 0 // no new view to report + } + + saveNewView := response != nil && response.View > newViewNum && response.Seq == latestDecisionSeq+1 + if saveNewView { + c.Logger.Infof("Node %d collected state with view %d and sequence %d", c.ID, response.View, response.Seq) + + // The record of the new view may truncate the WAL, so the current view has to be + // stopped first, and no other view may start until it is written. Otherwise a view + // could append a commit after that record and leave a commit without its + // pre-prepare in the WAL. + c.viewChangeLock.Lock() + c.abortCurrentView() + + // The fate of the in-flight proposal has to be decided before the record of the + // new view is written: the checkpoint alone does not say the in-flight proposal is + // no longer needed, maybePruneInFlight does. + if latestDecisionMetadata != nil { + c.maybePruneInFlight(latestDecisionMetadata) } - if response.View > newViewNum && response.Seq == latestDecisionSeq+1 { - c.Logger.Infof("Node %d collected state with view %d and sequence %d", c.ID, response.View, response.Seq) - newViewToSave := &protos.SavedMessage{ - Content: &protos.SavedMessage_NewView{ - NewView: &protos.ViewMetadata{ - ViewId: response.View, - LatestSequence: latestDecisionSeq, - DecisionsInView: 0, - }, + + newViewToSave := &protos.SavedMessage{ + Content: &protos.SavedMessage_NewView{ + NewView: &protos.ViewMetadata{ + ViewId: response.View, + LatestSequence: latestDecisionSeq, + DecisionsInView: 0, }, - } - if err := c.State.Save(newViewToSave); err != nil { - c.Logger.Panicf("Failed to save message to state, error: %v", err) - } - newViewNum = response.View - newDecisionsInView = 0 + }, + } + // The truncation flag is passed explicitly, after the view was stopped and the + // in-flight proposal was pruned, so it cannot change between this decision and + // the append. + truncateTo := c.InFlight.InFlightProposal() == nil + err := c.State.Save(newViewToSave, truncateTo) + c.viewChangeLock.Unlock() + + if err != nil { + c.Logger.Panicf("Failed to save message to state, error: %v", err) } - } - if latestDecisionMetadata != nil { + newViewNum = response.View + newDecisionsInView = 0 + } else if latestDecisionMetadata != nil { + // It has to be decided on every sync, not only when a new view record is saved, + // otherwise a stale proposal stays in flight and makes the node's view data + // invalid for the rest of the cluster. c.maybePruneInFlight(latestDecisionMetadata) } @@ -802,7 +953,6 @@ func (c *Controller) Start(startViewNumber uint64, startProposalSequence uint64, c.leaderToken = make(chan struct{}, 1) c.decisionChan = make(chan decision, 1) c.viewChange = make(chan viewInfo, 1) - c.abortViewChan = make(chan uint64, 1) Q, F := computeQuorum(c.N) c.Logger.Debugf("The number of nodes (N) is %d, F is %d, and the quorum size is %d", c.N, F, Q) @@ -890,18 +1040,25 @@ func (c *Controller) Decide(proposal types.Proposal, signatures []types.Signatur c.currViewLock.RUnlock() delivered := make(chan struct{}) - select { - case c.decisionChan <- decision{ + d := decision{ proposal: proposal, requests: requests, signatures: signatures, view: view, delivered: delivered, - }: + } + select { + case c.decisionChan <- d: case <-c.stopChan: // In case we are in the middle of shutting down, // abort deciding. return + case <-view.AbortChan(): + // The view is stopping while the controller is busy with a long sync or + // delivery. Do not block the view, or the abort would wait for it, but do + // not drop the decision either: hand it to the controller in the background. + go c.enqueueDecision(d) + return } select { @@ -911,6 +1068,16 @@ func (c *Controller) Decide(proposal types.Proposal, signatures []types.Signatur } } +// enqueueDecision hands a decision to the controller when the view that decided it is +// already stopping. The decision was accepted by a quorum, so it must not be lost: +// the controller delivers it once it is no longer busy. +func (c *Controller) enqueueDecision(d decision) { + select { + case c.decisionChan <- d: + case <-c.stopChan: + } +} + func (c *Controller) removeDeliveredFromPool(d decision) { for _, reqInfo := range d.requests { if err := c.RequestPool.RemoveRequest(reqInfo); err != nil { diff --git a/internal/bft/controller_sync_test.go b/internal/bft/controller_sync_test.go index dba98ee1..0b94cba2 100644 --- a/internal/bft/controller_sync_test.go +++ b/internal/bft/controller_sync_test.go @@ -6,13 +6,24 @@ package bft import ( + "errors" + "fmt" + "io" + "os" + "path/filepath" + "sync" "testing" "time" + "github.com/hyperledger/SmartBFT/pkg/api" + "github.com/hyperledger/SmartBFT/pkg/metrics/disabled" "github.com/hyperledger/SmartBFT/pkg/types" + "github.com/hyperledger/SmartBFT/pkg/wal" protos "github.com/hyperledger/SmartBFT/smartbftprotos" "github.com/stretchr/testify/require" "go.uber.org/zap" + "go.uber.org/zap/zapcore" + "google.golang.org/protobuf/proto" ) // stubSynchronizer returns a fixed SyncResponse. @@ -148,3 +159,474 @@ func TestSyncDecisionsInView(t *testing.T) { "DecisionsInView should be latestDecisionDecisions+1 when sync returns higher height") }) } + +// stubRequestsTimer is a no-op api.RequestsTimer. +type stubRequestsTimer struct{} + +func (s *stubRequestsTimer) StopTimers() {} +func (s *stubRequestsTimer) RestartTimers() {} +func (s *stubRequestsTimer) RemoveRequest(types.RequestInfo) error { + return nil +} + +// stateRespondingComm answers the state transfer request of sync() on behalf of +// every other node with the view and sequence the cluster claims to be at. +type stateRespondingComm struct { + stubComm + collector *StateCollector + view uint64 + seq uint64 +} + +func (c *stateRespondingComm) SendConsensus(targetID uint64, m *protos.Message) { + if m.GetStateTransferRequest() == nil { + return + } + + c.collector.HandleMessage(targetID, &protos.Message{ + Content: &protos.Message_StateTransferResponse{ + StateTransferResponse: &protos.StateTransferResponse{ + ViewNum: c.view, + Sequence: c.seq, + }, + }, + }) +} + +// syncFixture is a Controller whose sync() runs all the way to saving a record +// of the new view into a real WAL, together with the in-flight data that +// record's truncation flag is derived from. +type syncFixture struct { + controller *Controller + inFlight *InFlightData +} + +// newSyncFixture builds a controller that sits at controllerView and +// controllerSeq, and whose synchronizer returns a confirmed decision at syncSeq +// and syncView. The state transfer request is answered with a view one higher +// than syncView and syncSeq+1 as its sequence, which is what makes sync() save a +// record of the new view. +func newSyncFixture(t *testing.T, writeAheadLog api.WriteAheadLog, controllerSeq, controllerView, syncSeq, syncView uint64) *syncFixture { + t.Helper() + + log := newQuietLogger() + + checkpoint := &types.Checkpoint{} + checkpoint.Set(types.Proposal{ + Metadata: MarshalOrPanic(&protos.ViewMetadata{ + LatestSequence: controllerSeq, + ViewId: controllerView, + DecisionsInView: 1, + }), + }, []types.Signature{{ID: 1}, {ID: 2}, {ID: 3}}) + + collector := &StateCollector{ + SelfID: 1, + N: 4, + Logger: log, + CollectTimeout: 5 * time.Second, + } + collector.Start() + t.Cleanup(collector.Stop) + + viewChanger := &ViewChanger{ + SelfID: 1, + N: 4, + NodesList: []uint64{1, 2, 3, 4}, + Logger: log, + Comm: &stubComm{}, + RequestsTimer: &stubRequestsTimer{}, + Ticker: make(chan time.Time), + InMsqQSize: 10, + ControllerStartedWG: sync.WaitGroup{}, + MetricsViewChange: api.NewMetricsViewChange(&disabled.Provider{}), + } + viewChanger.Start(controllerView) + t.Cleanup(viewChanger.Stop) + + inFlight := &InFlightData{} + + controller := &Controller{ + ID: 1, + N: 4, + NodesList: []uint64{1, 2, 3, 4}, + Logger: log, + Comm: &stateRespondingComm{ + collector: collector, + view: syncView + 1, + seq: syncSeq + 1, + }, + Synchronizer: &stubSynchronizer{response: types.SyncResponse{ + Latest: types.Decision{ + Proposal: types.Proposal{ + Metadata: MarshalOrPanic(&protos.ViewMetadata{ + LatestSequence: syncSeq, + ViewId: syncView, + }), + }, + Signatures: []types.Signature{{ID: 1}, {ID: 2}, {ID: 3}}, + }, + Reconfig: types.ReconfigSync{InReplicatedDecisions: false}, + }}, + Checkpoint: checkpoint, + Collector: collector, + InFlight: inFlight, + ViewChanger: viewChanger, + State: &PersistedState{InFlightProposal: inFlight, Logger: log, WAL: writeAheadLog}, + WAL: writeAheadLog, + } + controller.syncChan = make(chan struct{}, 1) + + return &syncFixture{controller: controller, inFlight: inFlight} +} + +// newQuietLogger keeps the test output readable: the WAL and the controller both +// log a lot per record. +func newQuietLogger() api.Logger { + cfg := zap.NewDevelopmentConfig() + cfg.Level = zap.NewAtomicLevelAt(zapcore.ErrorLevel) + + logger, err := cfg.Build() + if err != nil { + panic(err) + } + + return logger.Sugar() +} + +// mustNoError fails the test right away when there is an error to report. +func mustNoError(t *testing.T, err error) { + t.Helper() + + if err != nil { + t.Fatalf("unexpected error: %v", err) + } +} + +// mustNotNil fails the test right away when there is nothing to look at. +func mustNotNil(t *testing.T, object any, message string) { + t.Helper() + + if object == nil { + t.Fatalf("%s", message) + } +} + +// storeInFlightProposal makes the controller hold a proposal at the given view +// and sequence, the way a node that proposed before losing the leader does. +func (f *syncFixture) storeInFlightProposal(view, seq uint64) { + f.inFlight.StoreProposal(types.Proposal{ + Metadata: MarshalOrPanic(&protos.ViewMetadata{ + ViewId: view, + LatestSequence: seq, + }), + }) +} + +// lastLogRecord returns the last record written into the WAL file with the +// given index. The CRC anchor is consumed by the reader and is not returned. +func lastLogRecord(t *testing.T, log api.Logger, dir string, index uint64) *protos.LogRecord { + t.Helper() + + reader, err := wal.NewLogRecordReader(log, filepath.Join(dir, fmt.Sprintf("%016x.wal", index))) + mustNoError(t, err) + defer func() { + require.NoError(t, reader.Close()) + }() + + var last *protos.LogRecord + for { + record, err := reader.Read() + if errors.Is(err, io.EOF) { + break + } + mustNoError(t, err) + + last = record + } + mustNotNil(t, last, "expected at least one record in the WAL") + + return last +} + +// walFileNames lists the files of the WAL directory. +func walFileNames(t *testing.T, dir string) []string { + t.Helper() + + entries, err := os.ReadDir(dir) + mustNoError(t, err) + + var names []string + for _, entry := range entries { + names = append(names, entry.Name()) + } + + return names +} + +type abortRecordingProposer struct { + aborted bool + abortChan chan struct{} +} + +func (p *abortRecordingProposer) Propose(types.Proposal) {} +func (p *abortRecordingProposer) Start() {} + +func (p *abortRecordingProposer) Abort() { + p.aborted = true + if p.abortChan != nil { + close(p.abortChan) + } +} + +func (p *abortRecordingProposer) Stopped() bool { return false } +func (p *abortRecordingProposer) AbortChan() <-chan struct{} { return p.abortChan } +func (p *abortRecordingProposer) GetLeaderID() uint64 { return 0 } +func (p *abortRecordingProposer) GetMetadata() []byte { return nil } +func (p *abortRecordingProposer) HandleMessage(uint64, *protos.Message) { +} + +type appendObservingWAL struct { + api.WriteAheadLog + observe func() +} + +func (w *appendObservingWAL) Append(entry []byte, truncateTo bool) error { + w.observe() + return w.WriteAheadLog.Append(entry, truncateTo) +} + +// TestSyncAbortsViewBeforeSavingNewView covers the write side of the truncation +// race: once the record of the new view is written, the view that was running +// when the sync started must not append anything after it. +func TestSyncAbortsViewBeforeSavingNewView(t *testing.T) { + const ( + controllerSeq = uint64(10) + controllerView = uint64(5) + inFlightSeq = uint64(10) + syncSeq = uint64(12) + syncView = uint64(6) + ) + + dir := t.TempDir() + log := newQuietLogger() + writeAheadLog, err := wal.Create(log, dir, nil) + mustNoError(t, err) + defer func() { + require.NoError(t, writeAheadLog.Close()) + }() + + proposer := &abortRecordingProposer{} + savedAfterAbort := true + observedWAL := &appendObservingWAL{ + WriteAheadLog: writeAheadLog, + observe: func() { + if !proposer.aborted { + savedAfterAbort = false + } + }, + } + + fixture := newSyncFixture(t, observedWAL, controllerSeq, controllerView, syncSeq, syncView) + fixture.storeInFlightProposal(controllerView, inFlightSeq) + fixture.controller.currView = proposer + + fixture.controller.sync() + + require.True(t, proposer.aborted, "the view was not aborted before the new view was saved") + require.True(t, savedAfterAbort, "the new view was saved before the view stopped") + + record := lastLogRecord(t, log, dir, 1) + saved := &protos.SavedMessage{} + mustNoError(t, proto.Unmarshal(record.GetData(), saved)) + mustNotNil(t, saved.GetNewView(), fmt.Sprintf("expected a record of the new view, got %v", saved)) +} + +// TestDecideDoesNotDropDecisionOnAbortedView covers the hand-off between the view +// and the controller: a view that is being aborted must not stay blocked on a full +// decision channel, but the committed decision must still reach the controller. +func TestDecideDoesNotDropDecisionOnAbortedView(t *testing.T) { + proposer := &abortRecordingProposer{abortChan: make(chan struct{})} + controller := &Controller{ + currView: proposer, + decisionChan: make(chan decision, 1), + stopChan: make(chan struct{}), + } + controller.decisionChan <- decision{proposal: types.Proposal{Payload: []byte("queued")}} + + decided := make(chan struct{}) + go func() { + controller.Decide(types.Proposal{Payload: []byte("committed")}, nil, nil) + close(decided) + }() + + select { + case <-decided: + t.Fatal("Decide returned before the view was aborted") + case <-time.After(50 * time.Millisecond): + } + + close(proposer.abortChan) + + select { + case <-decided: + case <-time.After(time.Second): + t.Fatal("Decide did not return after the view was aborted") + } + + queued := <-controller.decisionChan + require.Equal(t, []byte("queued"), queued.proposal.Payload) + + select { + case committed := <-controller.decisionChan: + require.Equal(t, []byte("committed"), committed.proposal.Payload) + case <-time.After(time.Second): + t.Fatal("the committed decision was dropped") + } +} + +// TestAbortViewDoesNotNeedTheControllerLoop covers the view changer side: aborting +// a view reaches the view directly, so it does not wait for the controller loop, +// which may be inside a long synchronization or an application delivery. +func TestAbortViewDoesNotNeedTheControllerLoop(t *testing.T) { + proposer := &guardTestProposer{} + controller := &Controller{ + currView: proposer, + currViewNumber: 1, + Batcher: guardTestBatcher{}, + Logger: newQuietLogger(), + } + + // No controller run loop is started: a blocking request handler would never answer. + require.True(t, controller.AbortView(1)) + require.True(t, proposer.Stopped()) +} + +// TestSyncSavesNewViewWithTruncateWhenInFlightIsPruned covers R01: a node that +// proposed sequence 10 and then caught up with a cluster that is at sequence 12 +// has nothing in flight once the sync is accounted for, so the record of the new +// view is the point from which everything older may go. +func TestSyncSavesNewViewWithTruncateWhenInFlightIsPruned(t *testing.T) { + const ( + controllerSeq = uint64(10) + controllerView = uint64(3) + inFlightSeq = uint64(10) + syncSeq = uint64(12) + syncView = uint64(4) + ) + + dir := t.TempDir() + log := newQuietLogger() + writeAheadLog, err := wal.Create(log, dir, nil) + mustNoError(t, err) + defer func() { + require.NoError(t, writeAheadLog.Close()) + }() + + fixture := newSyncFixture(t, writeAheadLog, controllerSeq, controllerView, syncSeq, syncView) + fixture.storeInFlightProposal(controllerView, inFlightSeq) + + fixture.controller.sync() + + require.Nil(t, fixture.inFlight.InFlightProposal(), "in-flight proposal at sequence %d is stale once the cluster is at %d", inFlightSeq, syncSeq) + + record := lastLogRecord(t, log, dir, 1) + require.True(t, record.GetTruncateTo(), "the record of the new view must be a truncation point, files: %v", walFileNames(t, dir)) + + saved := &protos.SavedMessage{} + mustNoError(t, proto.Unmarshal(record.GetData(), saved)) + newView := saved.GetNewView() + mustNotNil(t, newView, fmt.Sprintf("expected a record of the new view, got %v", saved)) + require.Equal(t, syncView+1, newView.GetViewId()) + require.Equal(t, syncSeq, newView.GetLatestSequence()) +} + +// TestSyncSavesNewViewWithoutTruncateWhenInFlightSurvives covers R01.1: the sync +// moved the checkpoint to sequence 12, but the node still holds a proposal at +// sequence 13, which the cluster has not seen. Losing the records before the new +// view would cost the node the pre-prepare of its own proposal. +func TestSyncSavesNewViewWithoutTruncateWhenInFlightSurvives(t *testing.T) { + const ( + controllerSeq = uint64(10) + controllerView = uint64(3) + inFlightSeq = uint64(13) + syncSeq = uint64(12) + syncView = uint64(4) + ) + + dir := t.TempDir() + log := newQuietLogger() + writeAheadLog, err := wal.Create(log, dir, nil) + mustNoError(t, err) + defer func() { + require.NoError(t, writeAheadLog.Close()) + }() + + fixture := newSyncFixture(t, writeAheadLog, controllerSeq, controllerView, syncSeq, syncView) + fixture.storeInFlightProposal(controllerView, inFlightSeq) + + fixture.controller.sync() + + inFlight := fixture.inFlight.InFlightProposal() + mustNotNil(t, inFlight, fmt.Sprintf("the in-flight proposal at sequence %d outlives a sync to %d", inFlightSeq, syncSeq)) + inFlightMD := &protos.ViewMetadata{} + mustNoError(t, proto.Unmarshal(inFlight.Metadata, inFlightMD)) + require.Equal(t, inFlightSeq, inFlightMD.GetLatestSequence()) + + record := lastLogRecord(t, log, dir, 1) + require.False(t, record.GetTruncateTo(), "a record written with a proposal in flight may not be a truncation point, files: %v", walFileNames(t, dir)) + + saved := &protos.SavedMessage{} + mustNoError(t, proto.Unmarshal(record.GetData(), saved)) + mustNotNil(t, saved.GetNewView(), fmt.Sprintf("expected a record of the new view, got %v", saved)) +} + +// TestSyncWALFileCountDoesNotGrowAcrossRestarts reproduces what an operator sees +// on a node that only ever syncs: every restart makes the WAL move to a new file +// and, absent a truncation point, nothing is ever freed. Each round the node +// holds a stale in-flight proposal and catches up with a cluster one sequence +// ahead, which is exactly the case where the record of the new view may truncate. +func TestSyncWALFileCountDoesNotGrowAcrossRestarts(t *testing.T) { + const ( + rounds = 5 + controllerView = uint64(3) + firstSyncSeq = uint64(12) + ) + + dir := t.TempDir() + log := newQuietLogger() + // A file size no record will ever reach, so the only rotations are the ones a + // restart makes on its own. + options := &wal.Options{FileSizeBytes: 64 * 1024 * 1024} + + writeAheadLog, err := wal.Create(log, dir, options) + mustNoError(t, err) + + var counts []int + for round := range rounds { + if round > 0 { + mustNoError(t, writeAheadLog.Close()) + + writeAheadLog, err = wal.Open(log, dir, options) + mustNoError(t, err) + + entries, err := writeAheadLog.ReadAll() + mustNoError(t, err) + require.Len(t, entries, 1, "only the record of the new view written by the previous round may be left to recover from") + } + defer func() { + require.NoError(t, writeAheadLog.Close()) + }() + + syncSeq := firstSyncSeq + uint64(round) + fixture := newSyncFixture(t, writeAheadLog, syncSeq-2, controllerView, syncSeq, controllerView+1) + // A proposal the cluster has already moved past. + fixture.storeInFlightProposal(controllerView, syncSeq-2) + + fixture.controller.sync() + + counts = append(counts, len(walFileNames(t, dir))) + } + + require.Equal(t, counts[1], counts[rounds-1], "the number of WAL files must stop growing after the first restart, got %v", counts) +} diff --git a/internal/bft/controller_test.go b/internal/bft/controller_test.go index 548099eb..fae2e8f6 100644 --- a/internal/bft/controller_test.go +++ b/internal/bft/controller_test.go @@ -76,6 +76,99 @@ func TestControllerBasic(t *testing.T) { controller.Stop() } +type blockingProposer struct { + abortStarted chan struct{} + abortRelease chan struct{} + once sync.Once +} + +func (p *blockingProposer) Propose(types.Proposal) {} +func (p *blockingProposer) Start() {} + +func (p *blockingProposer) Abort() { + p.once.Do(func() { + close(p.abortStarted) + }) + <-p.abortRelease +} + +func (p *blockingProposer) Stopped() bool { return false } +func (p *blockingProposer) AbortChan() <-chan struct{} { return nil } +func (p *blockingProposer) GetLeaderID() uint64 { return 0 } +func (p *blockingProposer) GetMetadata() []byte { return nil } +func (p *blockingProposer) HandleMessage(uint64, *protos.Message) {} + +type fixedProposerBuilder struct { + proposer bft.Proposer +} + +func (b *fixedProposerBuilder) NewProposer(uint64, uint64, uint64, uint64, int) (bft.Proposer, bft.Phase) { + return b.proposer, bft.COMMITTED +} + +func TestAbortViewWaitsUntilTheViewStops(t *testing.T) { + log := zap.NewNop().Sugar() + batcher := &mocks.Batcher{} + batcher.On("Close") + pool := &mocks.RequestPool{} + pool.On("Close") + leaderMon := &mocks.LeaderMonitor{} + leaderMon.On("ChangeRole", mock.Anything, mock.Anything, mock.Anything) + leaderMon.On("Close") + + proposer := &blockingProposer{ + abortStarted: make(chan struct{}), + abortRelease: make(chan struct{}), + } + + startedWG := sync.WaitGroup{} + startedWG.Add(1) + + controller := &bft.Controller{ + Checkpoint: &types.Checkpoint{}, + Batcher: batcher, + RequestPool: pool, + LeaderMonitor: leaderMon, + ProposerBuilder: &fixedProposerBuilder{proposer: proposer}, + ID: 0, + N: 4, + NodesList: []uint64{0, 1, 2, 3}, + Logger: log, + StartedWG: &startedWG, + ViewSequences: &atomic.Value{}, + } + + controller.Start(1, 0, 0, 1, false) + + abortReturned := make(chan struct{}) + go func() { + controller.AbortView(1) + close(abortReturned) + }() + + select { + case <-proposer.abortStarted: + case <-time.After(time.Second): + require.Fail(t, "the view was not aborted") + } + + select { + case <-abortReturned: + require.Fail(t, "AbortView returned before the view stopped") + case <-time.After(100 * time.Millisecond): + } + + close(proposer.abortRelease) + + select { + case <-abortReturned: + case <-time.After(time.Second): + require.Fail(t, "AbortView did not return after the view stopped") + } + + controller.Stop() +} + func TestControllerLeaderBasic(t *testing.T) { basicLog, err := zap.NewDevelopment() require.NoError(t, err) @@ -775,7 +868,7 @@ func TestSyncInform(t *testing.T) { reqTimer.On("StopTimers") reqTimer.On("RestartTimers") controllerMock := &mocks.ViewController{} - controllerMock.On("AbortView", mock.Anything) + controllerMock.On("AbortView", mock.Anything).Return(true) collector := bft.StateCollector{ SelfID: 0, @@ -1222,7 +1315,7 @@ func TestDeliverTwiceOnceFromSyncAndOnceFromViewData(t *testing.T) { batcher.On("Close") state := &mocks.State{} - state.On("Save", mock.Anything).Return(nil) + state.On("Save", mock.Anything, mock.Anything).Return(nil) verifier := &mocks.VerifierMock{} verifier.On("VerificationSequence").Return(uint64(1)) diff --git a/internal/bft/controller_view_guard_test.go b/internal/bft/controller_view_guard_test.go new file mode 100644 index 00000000..d9630c63 --- /dev/null +++ b/internal/bft/controller_view_guard_test.go @@ -0,0 +1,156 @@ +// Copyright IBM Corp. All Rights Reserved. +// +// SPDX-License-Identifier: Apache-2.0 +// + +package bft + +import ( + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/hyperledger/SmartBFT/pkg/types" + protos "github.com/hyperledger/SmartBFT/smartbftprotos" + "github.com/stretchr/testify/require" +) + +type guardTestBatcher struct{} + +func (guardTestBatcher) NextBatch() [][]byte { return nil } +func (guardTestBatcher) Close() {} +func (guardTestBatcher) Closed() bool { return false } +func (guardTestBatcher) Reset() {} + +type guardTestRequestPool struct{} + +func (guardTestRequestPool) Prune(func([]byte) error) {} +func (guardTestRequestPool) Submit([]byte) error { return nil } +func (guardTestRequestPool) Size() int { return 0 } +func (guardTestRequestPool) NextRequests(int, uint64, bool) ([][]byte, bool) { + return nil, false +} +func (guardTestRequestPool) RemoveRequest(types.RequestInfo) error { return nil } +func (guardTestRequestPool) StopTimers() {} +func (guardTestRequestPool) RestartTimers() {} +func (guardTestRequestPool) Close() {} + +type guardTestLeaderMonitor struct{} + +func (guardTestLeaderMonitor) ChangeRole(Role, uint64, uint64) {} +func (guardTestLeaderMonitor) ProcessMsg(uint64, *protos.Message) {} +func (guardTestLeaderMonitor) InjectArtificialHeartbeat(uint64, *protos.Message) {} +func (guardTestLeaderMonitor) HeartbeatWasSent() {} +func (guardTestLeaderMonitor) Close() {} +func (guardTestLeaderMonitor) StopLeaderSendMsg() {} + +type guardTestProposer struct { + aborted atomic.Bool +} + +func (p *guardTestProposer) Propose(types.Proposal) {} +func (p *guardTestProposer) Start() {} +func (p *guardTestProposer) Abort() { p.aborted.Store(true) } +func (p *guardTestProposer) Stopped() bool { return p.aborted.Load() } +func (p *guardTestProposer) AbortChan() <-chan struct{} { return nil } +func (p *guardTestProposer) GetLeaderID() uint64 { return 0 } +func (p *guardTestProposer) GetMetadata() []byte { return nil } +func (p *guardTestProposer) HandleMessage(uint64, *protos.Message) {} + +type guardTestProposerBuilder struct { + starts atomic.Int64 +} + +func (b *guardTestProposerBuilder) NewProposer(uint64, uint64, uint64, uint64, int) (Proposer, Phase) { + b.starts.Add(1) + return &guardTestProposer{}, COMMITTED +} + +func TestChangeViewDoesNotRestartViewLeftByViewChanger(t *testing.T) { + builder := &guardTestProposerBuilder{} + + startedWG := sync.WaitGroup{} + startedWG.Add(1) + + controller := &Controller{ + Checkpoint: &types.Checkpoint{}, + Batcher: guardTestBatcher{}, + RequestPool: guardTestRequestPool{}, + LeaderMonitor: guardTestLeaderMonitor{}, + ProposerBuilder: builder, + ID: 0, + N: 4, + NodesList: []uint64{0, 1, 2, 3}, + Logger: newQuietLogger(), + StartedWG: &startedWG, + } + + controller.Start(1, 0, 0, 0, false) + defer controller.Stop() + + require.Equal(t, int64(1), builder.starts.Load()) + + controller.holdView(1) + require.True(t, controller.isViewHeld(1)) + + controller.changeView(1, 1, 0) + + require.Equal(t, int64(1), builder.starts.Load(), "the controller restarted the view the view changer left") + + controller.releaseHeldView() + require.False(t, controller.isViewHeld(1)) + + controller.changeView(1, 1, 0) + + require.Equal(t, int64(2), builder.starts.Load(), "the view was not restarted after the hold was released") +} + +func TestAbortViewAndHoldBlocksViewChangesUntilReleased(t *testing.T) { + builder := &guardTestProposerBuilder{} + + startedWG := sync.WaitGroup{} + startedWG.Add(1) + + controller := &Controller{ + Checkpoint: &types.Checkpoint{}, + Batcher: guardTestBatcher{}, + RequestPool: guardTestRequestPool{}, + LeaderMonitor: guardTestLeaderMonitor{}, + ProposerBuilder: builder, + ID: 0, + N: 4, + NodesList: []uint64{0, 1, 2, 3}, + Logger: newQuietLogger(), + StartedWG: &startedWG, + } + + controller.Start(1, 0, 0, 0, false) + defer controller.Stop() + + require.True(t, controller.AbortViewAndHold(1)) + + changed := make(chan struct{}) + go func() { + controller.changeView(1, 1, 0) + close(changed) + }() + + select { + case <-changed: + t.Fatal("changeView returned while the view changer was writing its record") + case <-time.After(50 * time.Millisecond): + } + + require.Equal(t, int64(1), builder.starts.Load(), "the view was restarted before the view changer released it") + + controller.ReleaseViewWrite() + + select { + case <-changed: + case <-time.After(time.Second): + t.Fatal("changeView did not continue after the view changer released the write") + } + + require.Equal(t, int64(2), builder.starts.Load()) +} diff --git a/internal/bft/mocks/state.go b/internal/bft/mocks/state.go index f8a9874a..8c327990 100644 --- a/internal/bft/mocks/state.go +++ b/internal/bft/mocks/state.go @@ -98,16 +98,16 @@ func (_c *State_Restore_Call) RunAndReturn(run func(view *bft.View) error) *Stat } // Save provides a mock function for the type State -func (_mock *State) Save(message *smartbftprotos.SavedMessage) error { - ret := _mock.Called(message) +func (_mock *State) Save(message *smartbftprotos.SavedMessage, truncateTo bool) error { + ret := _mock.Called(message, truncateTo) if len(ret) == 0 { panic("no return value specified for Save") } var r0 error - if returnFunc, ok := ret.Get(0).(func(*smartbftprotos.SavedMessage) error); ok { - r0 = returnFunc(message) + if returnFunc, ok := ret.Get(0).(func(*smartbftprotos.SavedMessage, bool) error); ok { + r0 = returnFunc(message, truncateTo) } else { r0 = ret.Error(0) } @@ -121,18 +121,24 @@ type State_Save_Call struct { // Save is a helper method to define mock.On call // - message *smartbftprotos.SavedMessage -func (_e *State_Expecter) Save(message any) *State_Save_Call { - return &State_Save_Call{Call: _e.mock.On("Save", message)} +// - truncateTo bool +func (_e *State_Expecter) Save(message any, truncateTo any) *State_Save_Call { + return &State_Save_Call{Call: _e.mock.On("Save", message, truncateTo)} } -func (_c *State_Save_Call) Run(run func(message *smartbftprotos.SavedMessage)) *State_Save_Call { +func (_c *State_Save_Call) Run(run func(message *smartbftprotos.SavedMessage, truncateTo bool)) *State_Save_Call { _c.Call.Run(func(args mock.Arguments) { var arg0 *smartbftprotos.SavedMessage if args[0] != nil { arg0 = args[0].(*smartbftprotos.SavedMessage) } + var arg1 bool + if args[1] != nil { + arg1 = args[1].(bool) + } run( arg0, + arg1, ) }) return _c @@ -143,7 +149,7 @@ func (_c *State_Save_Call) Return(err error) *State_Save_Call { return _c } -func (_c *State_Save_Call) RunAndReturn(run func(message *smartbftprotos.SavedMessage) error) *State_Save_Call { +func (_c *State_Save_Call) RunAndReturn(run func(message *smartbftprotos.SavedMessage, truncateTo bool) error) *State_Save_Call { _c.Call.Return(run) return _c } diff --git a/internal/bft/mocks/view_controller.go b/internal/bft/mocks/view_controller.go index c3932c30..9e9a0a6e 100644 --- a/internal/bft/mocks/view_controller.go +++ b/internal/bft/mocks/view_controller.go @@ -45,9 +45,20 @@ func (_m *ViewController) EXPECT() *ViewController_Expecter { } // AbortView provides a mock function for the type ViewController -func (_mock *ViewController) AbortView(view uint64) { - _mock.Called(view) - return +func (_mock *ViewController) AbortView(view uint64) bool { + ret := _mock.Called(view) + + if len(ret) == 0 { + panic("no return value specified for AbortView") + } + + var r0 bool + if returnFunc, ok := ret.Get(0).(func(uint64) bool); ok { + r0 = returnFunc(view) + } else { + r0 = ret.Get(0).(bool) + } + return r0 } // ViewController_AbortView_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'AbortView' @@ -74,13 +85,13 @@ func (_c *ViewController_AbortView_Call) Run(run func(view uint64)) *ViewControl return _c } -func (_c *ViewController_AbortView_Call) Return() *ViewController_AbortView_Call { - _c.Call.Return() +func (_c *ViewController_AbortView_Call) Return(b bool) *ViewController_AbortView_Call { + _c.Call.Return(b) return _c } -func (_c *ViewController_AbortView_Call) RunAndReturn(run func(view uint64)) *ViewController_AbortView_Call { - _c.Run(run) +func (_c *ViewController_AbortView_Call) RunAndReturn(run func(view uint64) bool) *ViewController_AbortView_Call { + _c.Call.Return(run) return _c } diff --git a/internal/bft/state.go b/internal/bft/state.go index 3e79de17..2befa9c9 100644 --- a/internal/bft/state.go +++ b/internal/bft/state.go @@ -20,7 +20,7 @@ type StateRecorder struct { SavedMessages []*protos.SavedMessage } -func (sr *StateRecorder) Save(message *protos.SavedMessage) error { +func (sr *StateRecorder) Save(message *protos.SavedMessage, _ bool) error { sr.SavedMessages = append(sr.SavedMessages, message) return nil } @@ -36,7 +36,7 @@ type PersistedState struct { WAL api.WriteAheadLog } -func (ps *PersistedState) Save(msgToSave *protos.SavedMessage) error { +func (ps *PersistedState) Save(msgToSave *protos.SavedMessage, truncateTo bool) error { if proposed := msgToSave.GetProposedRecord(); proposed != nil { ps.storeProposal(proposed) } @@ -48,15 +48,13 @@ func (ps *PersistedState) Save(msgToSave *protos.SavedMessage) error { if err != nil { ps.Logger.Panicf("Failed marshaling message: %v", err) } - // It is only safe to truncate if we either: - // - // 1) Process a pre-prepare, because it means we safely persisted the - // previous proposal and have a stable checkpoint. - // 2) Acquired a new view message which contains 2f+1 attestations - // of the cluster agreeing to a new view configuration. - newProposal := msgToSave.GetProposedRecord() != nil - // TODO: handle view message here as well, and add "|| finalizedView" to truncate flag - return ps.WAL.Append(b, newProposal) + + // The caller decides whether this record is a truncation point. It knows + // whether the node still holds a proposal that recovery needs, and whether + // the view that could append after this record has stopped. Computing the + // flag here from the shared in-flight state would make it depend on which + // goroutine happens to change that state first. + return ps.WAL.Append(b, truncateTo) } func (ps *PersistedState) storeProposal(proposed *protos.ProposedRecord) { diff --git a/internal/bft/state_test.go b/internal/bft/state_test.go index 948fa3b2..4713cc6a 100644 --- a/internal/bft/state_test.go +++ b/internal/bft/state_test.go @@ -3,29 +3,38 @@ // SPDX-License-Identifier: Apache-2.0 // -package bft_test +package bft import ( + "os" + "strings" "testing" - "github.com/hyperledger/SmartBFT/internal/bft" - "github.com/hyperledger/SmartBFT/internal/bft/mocks" "github.com/hyperledger/SmartBFT/pkg/api" "github.com/hyperledger/SmartBFT/pkg/metrics/disabled" "github.com/hyperledger/SmartBFT/pkg/types" + "github.com/hyperledger/SmartBFT/pkg/wal" protos "github.com/hyperledger/SmartBFT/smartbftprotos" - "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" "go.uber.org/zap" "google.golang.org/protobuf/proto" ) +// requestsFromProposalVerifier returns no requests, which is all TestStateRestore needs from a verifier. +type requestsFromProposalVerifier struct { + api.Verifier +} + +func (v *requestsFromProposalVerifier) RequestsFromProposal(types.Proposal) []types.RequestInfo { + return nil +} + func TestStateRestore(t *testing.T) { prePrepare := &protos.PrePrepare{ Proposal: &protos.Proposal{ Header: []byte{1}, Payload: []byte{1}, - Metadata: bft.MarshalOrPanic(&protos.ViewMetadata{ + Metadata: MarshalOrPanic(&protos.ViewMetadata{ DecisionsInView: 0, LatestSequence: 0, ViewId: 1, @@ -76,7 +85,7 @@ func TestStateRestore(t *testing.T) { description string WALContent [][]byte expectedError string - expectedPhase bft.Phase + expectedPhase Phase expectedViewNumber uint64 expectedProposalSeq uint64 expectedInFlightProposal *types.Proposal @@ -99,74 +108,74 @@ func TestStateRestore(t *testing.T) { }, { description: "proposed", - expectedPhase: bft.PROPOSED, + expectedPhase: PROPOSED, expectedViewNumber: 300, expectedProposalSeq: 200, - WALContent: [][]byte{bft.MarshalOrPanic(proposedRecord)}, + WALContent: [][]byte{MarshalOrPanic(proposedRecord)}, expectedInFlightProposal: expectedInFlightProposal, }, { description: "commit persisted but pre-prepare nowhere to be found", - expectedPhase: bft.PREPARED, + expectedPhase: PREPARED, WALContent: [][]byte{ - bft.MarshalOrPanic(preparedProof), + MarshalOrPanic(preparedProof), }, expectedError: "last message is a commit, but expected to also have a matching pre-prepare", }, { description: "prepared but not committed", - expectedPhase: bft.PREPARED, + expectedPhase: PREPARED, expectedViewNumber: 300, expectedProposalSeq: 200, // Metadata holds sequence 199, WAL has sequence 200 proposalSeqViewInitializedWith: 200, WALContent: [][]byte{ - bft.MarshalOrPanic(proposedRecord), - bft.MarshalOrPanic(preparedProof), + MarshalOrPanic(proposedRecord), + MarshalOrPanic(preparedProof), }, expectedInFlightProposal: expectedInFlightProposal, expectedInFlightPrepared: true, }, { description: "prepared and committed", - expectedPhase: bft.COMMITTED, + expectedPhase: COMMITTED, expectedViewNumber: 300, expectedProposalSeq: 201, // Metadata holds sequence 200 which is the same as WAL proposalSeqViewInitializedWith: 201, WALContent: [][]byte{ - bft.MarshalOrPanic(proposedRecord), - bft.MarshalOrPanic(preparedProof), + MarshalOrPanic(proposedRecord), + MarshalOrPanic(preparedProof), }, }, { description: "WAL out of sync", - expectedPhase: bft.COMMITTED, + expectedPhase: COMMITTED, // Metadata holds sequence 198 but WAL has sequence 200 proposalSeqViewInitializedWith: 199, WALContent: [][]byte{ - bft.MarshalOrPanic(proposedRecord), - bft.MarshalOrPanic(preparedProof), + MarshalOrPanic(proposedRecord), + MarshalOrPanic(preparedProof), }, expectedError: "last proposal sequence persisted into WAL is 200" + " which is greater than last committed sequence is 199", }, { description: "malformed prepared but valid commit", - expectedPhase: bft.COMMITTED, + expectedPhase: COMMITTED, WALContent: [][]byte{ {1, 2, 3}, - bft.MarshalOrPanic(preparedProof), + MarshalOrPanic(preparedProof), }, expectedError: "failed unmarshaling last entry from WAL: " + "proto:", }, { description: "malformed prepared but empty commit", - expectedPhase: bft.COMMITTED, + expectedPhase: COMMITTED, WALContent: [][]byte{ - bft.MarshalOrPanic(&protos.SavedMessage{Content: nil}), - bft.MarshalOrPanic(preparedProof), + MarshalOrPanic(&protos.SavedMessage{Content: nil}), + MarshalOrPanic(preparedProof), }, expectedError: "expected second last message to be a pre-prepare, but got '' instead", }, @@ -176,15 +185,14 @@ func TestStateRestore(t *testing.T) { require.NoError(t, err) log := basicLog.Sugar() - state := &bft.PersistedState{ + state := &PersistedState{ Entries: testCase.WALContent, Logger: log, - InFlightProposal: &bft.InFlightData{}, + InFlightProposal: &InFlightData{}, } - verifier := &mocks.VerifierMock{} - verifier.On("RequestsFromProposal", mock.Anything).Return(nil) - view := &bft.View{ + verifier := &requestsFromProposalVerifier{} + view := &View{ Number: 300, ProposalSequence: testCase.proposalSeqViewInitializedWith, MetricsView: api.NewMetricsView(&disabled.Provider{}), @@ -252,16 +260,16 @@ func TestStateLoadIfApplicable(t *testing.T) { }, { description: "last record is a proposal", - WALContent: [][]byte{bft.MarshalOrPanic(newView), bft.MarshalOrPanic(proposedRecord)}, + WALContent: [][]byte{MarshalOrPanic(newView), MarshalOrPanic(proposedRecord)}, }, { description: "last record is a new view", - WALContent: [][]byte{bft.MarshalOrPanic(savedViewChange), bft.MarshalOrPanic(newView)}, + WALContent: [][]byte{MarshalOrPanic(savedViewChange), MarshalOrPanic(newView)}, expectedViewAndSeq: &types.ViewAndSeq{View: 5, Seq: 10}, }, { description: "last record is a view change", - WALContent: [][]byte{bft.MarshalOrPanic(newView), bft.MarshalOrPanic(savedViewChange)}, + WALContent: [][]byte{MarshalOrPanic(newView), MarshalOrPanic(savedViewChange)}, expectedViewChange: viewChange, }, } { @@ -269,7 +277,7 @@ func TestStateLoadIfApplicable(t *testing.T) { basicLog, err := zap.NewDevelopment() require.NoError(t, err) - state := &bft.PersistedState{ + state := &PersistedState{ Entries: testCase.WALContent, Logger: basicLog.Sugar(), } @@ -296,3 +304,227 @@ func TestStateLoadIfApplicable(t *testing.T) { }) } } + +// 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 { + appended [][]byte + truncateTo []bool +} + +func (w *recordingWAL) Append(entry []byte, truncateTo bool) error { + w.appended = append(w.appended, entry) + w.truncateTo = append(w.truncateTo, truncateTo) + + return nil +} + +func newViewRecord(viewID, latestSequence uint64) *protos.SavedMessage { + return &protos.SavedMessage{ + Content: &protos.SavedMessage_NewView{ + NewView: &protos.ViewMetadata{ + ViewId: viewID, + LatestSequence: latestSequence, + }, + }, + } +} + +// The caller decides where the WAL may be truncated. Save must persist that +// decision as given, together with the record. +func TestSaveUsesTheGivenTruncationFlag(t *testing.T) { + proposalRecord := &protos.SavedMessage{ + Content: &protos.SavedMessage_ProposedRecord{ + ProposedRecord: &protos.ProposedRecord{ + PrePrepare: &protos.PrePrepare{ + Proposal: &protos.Proposal{ + Header: []byte{1}, + Payload: []byte{1}, + Metadata: MarshalOrPanic(&protos.ViewMetadata{ViewId: 1}), + VerificationSequence: 100, + }, + Seq: 200, + View: 300, + }, + Prepare: &protos.Prepare{ + Seq: 200, + View: 300, + }, + }, + }, + } + + commitRecord := &protos.SavedMessage{ + Content: &protos.SavedMessage_Commit{ + Commit: &protos.Message{ + Content: &protos.Message_Commit{ + Commit: &protos.Commit{ + Seq: 200, + View: 300, + Signature: &protos.Signature{ + Signer: 11, + }, + }, + }, + }, + }, + } + + for _, testCase := range []struct { + description string + proposalInFlight bool + recordsToSave []*protos.SavedMessage + truncateTo []bool + }{ + { + description: "a proposal begins the pair, so it is a truncation point", + recordsToSave: []*protos.SavedMessage{proposalRecord}, + truncateTo: []bool{true}, + }, + { + description: "a commit right after the proposal keeps the pair", + recordsToSave: []*protos.SavedMessage{proposalRecord, commitRecord}, + truncateTo: []bool{true, false}, + }, + { + description: "a new view with nothing in flight is a truncation point", + recordsToSave: []*protos.SavedMessage{newViewRecord(7, 4)}, + truncateTo: []bool{true}, + }, + { + description: "a new view with a proposal in flight is not a truncation point", + proposalInFlight: true, + recordsToSave: []*protos.SavedMessage{newViewRecord(7, 4)}, + truncateTo: []bool{false}, + }, + } { + t.Run(testCase.description, func(t *testing.T) { + basicLog, err := zap.NewDevelopment() + require.NoError(t, err) + + inFlight := &InFlightData{} + if testCase.proposalInFlight { + inFlight.StoreProposal(types.Proposal{VerificationSequence: 42}) + } + + writeAheadLog := &recordingWAL{} + state := &PersistedState{ + InFlightProposal: inFlight, + Logger: basicLog.Sugar(), + WAL: writeAheadLog, + } + + for i, record := range testCase.recordsToSave { + require.NoError(t, state.Save(record, testCase.truncateTo[i])) + } + + require.Equal(t, testCase.truncateTo, writeAheadLog.truncateTo) + + require.Len(t, writeAheadLog.appended, len(testCase.recordsToSave)) + for i, record := range testCase.recordsToSave { + appended := &protos.SavedMessage{} + require.NoError(t, proto.Unmarshal(writeAheadLog.appended[i], appended)) + require.True(t, proto.Equal(record, appended), "record %d was not appended as saved", i) + } + }) + } +} + +func walFileCount(t *testing.T, dirPath string) int { + t.Helper() + + names, err := os.ReadDir(dirPath) + require.NoError(t, err) + + var count int + for _, name := range names { + if strings.HasSuffix(name.Name(), ".wal") { + count++ + } + } + + return count +} + +// A node that syncs but never proposes writes new view records only. Each of them +// is a truncation point, so rotations drop the files written before it, and neither +// another run of writes nor a restart adds files. +func TestSaveNewViewWithNoProposalInFlightDoesNotGrowWAL(t *testing.T) { + basicLog, err := zap.NewDevelopment() + require.NoError(t, err) + log := basicLog.Sugar() + + testDir := t.TempDir() + + options := &wal.Options{FileSizeBytes: 1024} + writeAheadLog, err := wal.Create(log, testDir, options) + require.NoError(t, err) + + state := &PersistedState{ + InFlightProposal: &InFlightData{}, + Logger: log, + WAL: writeAheadLog, + } + + for i := range 100 { + require.NoError(t, state.Save(newViewRecord(7, uint64(i)), true)) + } + require.NoError(t, writeAheadLog.Close()) + + filesBeforeRestart := walFileCount(t, testDir) + require.Positive(t, filesBeforeRestart) + + for restart := range 2 { + writeAheadLog, err = wal.Open(log, testDir, options) + require.NoError(t, err) + state.WAL = writeAheadLog + + entries, err := writeAheadLog.ReadAll() + require.NoError(t, err) + require.NotEmpty(t, entries) + + for i := range 100 { + require.NoError(t, state.Save(newViewRecord(8, uint64(i)), true)) + } + require.NoError(t, writeAheadLog.Close()) + + require.LessOrEqual(t, walFileCount(t, testDir), filesBeforeRestart, + "restart %d grew the WAL", restart) + } +} + +// A commit is never saved by the view with truncate=true: its pre-prepare has to +// survive next to it. The flag is not read from the shared in-flight state, so it +// cannot change while Save is running. +func TestSaveCommitUsesTheGivenTruncationFlag(t *testing.T) { + basicLog, err := zap.NewDevelopment() + require.NoError(t, err) + + inFlight := &InFlightData{} + inFlight.StoreProposal(types.Proposal{VerificationSequence: 42}) + + commitRecord := &protos.SavedMessage{ + Content: &protos.SavedMessage_Commit{ + Commit: &protos.Message{ + Content: &protos.Message_Commit{ + Commit: &protos.Commit{ + Seq: 200, + View: 300, + }, + }, + }, + }, + } + + writeAheadLog := &recordingWAL{} + state := &PersistedState{ + InFlightProposal: inFlight, + Logger: basicLog.Sugar(), + WAL: writeAheadLog, + } + + require.NoError(t, state.Save(commitRecord, false)) + + require.Equal(t, []bool{false}, writeAheadLog.truncateTo, "a commit must never be a truncation point") + require.True(t, inFlight.IsInFlightPrepared(), "saving a commit must not clear the proposal in flight") +} diff --git a/internal/bft/view.go b/internal/bft/view.go index e9f7bef6..07fd1e88 100644 --- a/internal/bft/view.go +++ b/internal/bft/view.go @@ -48,8 +48,10 @@ func (p Phase) String() string { // State can save and restore the state type State interface { - // Save saves a message. - Save(message *protos.SavedMessage) error + // Save saves a message. truncateTo tells the WAL whether everything written + // before the message can be dropped. The caller decides it, because only the + // caller knows whether the node still has a proposal to recover. + Save(message *protos.SavedMessage, truncateTo bool) error // Restore restores the given view to its latest state // before a crash, if applicable. @@ -399,7 +401,8 @@ func (v *View) processProposal() Phase { prepareMessage := v.createPrepare(seq, proposal) // We are about to send a prepare for a pre-prepare, - // so we record the pre-prepare. + // so we record the pre-prepare. The record starts a new proposal pair, + // hence everything written before it may be truncated. savedMsg := &protos.SavedMessage{ Content: &protos.SavedMessage_ProposedRecord{ ProposedRecord: &protos.ProposedRecord{ @@ -408,7 +411,7 @@ func (v *View) processProposal() Phase { }, }, } - if err = v.State.Save(savedMsg); err != nil { + if err = v.State.Save(savedMsg, true); err != nil { v.Logger.Panicf("Failed to save message to state, error: %v", err) } v.lastBroadcastSent = prepareMessage @@ -503,8 +506,9 @@ func (v *View) processPrepares() Phase { } // We received enough prepares to send a commit. - // Save the commit message we are about to send. - if err = v.State.Save(preparedProof); err != nil { + // Save the commit message we are about to send. A commit must not be a + // truncation point: its pre-prepare has to survive next to it. + if err = v.State.Save(preparedProof, false); err != nil { v.Logger.Panicf("Failed to save message to state, error: %v", err) } v.currCommitSent = proto.Clone(commitMsg).(*protos.Message) diff --git a/internal/bft/view_test.go b/internal/bft/view_test.go index e7f6e02b..f19a6854 100644 --- a/internal/bft/view_test.go +++ b/internal/bft/view_test.go @@ -8,6 +8,7 @@ package bft_test import ( "errors" "fmt" + "os" "strings" "sync" "sync/atomic" @@ -1005,9 +1006,9 @@ func TestViewPersisted(t *testing.T) { } var persistedToLog sync.WaitGroup persistedToLog.Add(1) - state.On("Save", mock.Anything).Return(nil).Run(func(args mock.Arguments) { + state.On("Save", mock.Anything, mock.Anything).Return(nil).Run(func(args mock.Arguments) { persistedToLog.Done() - persistedState.Save(args.Get(0).(*protos.SavedMessage)) + persistedState.Save(args.Get(0).(*protos.SavedMessage), args.Get(1).(bool)) }) view.Start() @@ -1539,7 +1540,7 @@ func newView(t *testing.T, selfID uint64, network map[uint64]*testedView) *teste }) state := &mocks.State{} - state.On("Save", mock.Anything).Return(nil) + state.On("Save", mock.Anything, mock.Anything).Return(nil) tv.View = &bft.View{ RetrieveCheckpoint: (&types.Checkpoint{}).Get, @@ -1563,3 +1564,116 @@ func newView(t *testing.T, selfID uint64, network map[uint64]*testedView) *teste return tv } + +// logFileCount returns the number of WAL files in the given directory. +func logFileCount(t *testing.T, dirPath string) int { + t.Helper() + + names, err := os.ReadDir(dirPath) + require.NoError(t, err) + + var count int + for _, name := range names { + if strings.HasSuffix(name.Name(), ".wal") { + count++ + } + } + + return count +} + +// TestSaveNewViewAfterConfirmedSyncTruncatesLog reproduces R07 at the +// PersistedState seam, with a real WAL: once a sync has confirmed the proposal +// with a quorum, the in-flight data is pruned and the record of the new view is +// the truncation point. A restart then reads back the bare record of the new +// view, and the number of log files does not grow with the restart. +func TestSaveNewViewAfterConfirmedSyncTruncatesLog(t *testing.T) { + basicLog, err := zap.NewDevelopment() + require.NoError(t, err) + log := basicLog.Sugar() + + testDir := t.TempDir() + + options := &wal.Options{FileSizeBytes: 4096} + writeAheadLog, err := wal.Create(log, testDir, options) + require.NoError(t, err) + + proposedRecord := &protos.SavedMessage{ + Content: &protos.SavedMessage_ProposedRecord{ + ProposedRecord: &protos.ProposedRecord{ + PrePrepare: &protos.PrePrepare{ + Proposal: &protos.Proposal{ + Header: []byte{1}, + Payload: []byte{2}, + Metadata: bft.MarshalOrPanic(&protos.ViewMetadata{ViewId: 3, LatestSequence: 4}), + VerificationSequence: 5, + }, + Seq: 5, + View: 3, + }, + Prepare: &protos.Prepare{ + Seq: 5, + View: 3, + }, + }, + }, + } + + commitRecord := &protos.SavedMessage{ + Content: &protos.SavedMessage_Commit{ + Commit: &protos.Message{ + Content: &protos.Message_Commit{ + Commit: &protos.Commit{ + Seq: 5, + View: 3, + }, + }, + }, + }, + } + + newViewRecord := &protos.SavedMessage{ + Content: &protos.SavedMessage_NewView{ + NewView: &protos.ViewMetadata{ + ViewId: 4, + LatestSequence: 5, + }, + }, + } + + state := &bft.PersistedState{ + InFlightProposal: &bft.InFlightData{}, + Logger: log, + WAL: writeAheadLog, + } + + // The pair of a proposal in flight has to survive on its own. + require.NoError(t, state.Save(proposedRecord, true)) + require.NoError(t, state.Save(commitRecord, false)) + + // The sync confirmed the proposal with a quorum, so the in-flight data is + // stale, and the record of the new view becomes the truncation point. + state.InFlightProposal = &bft.InFlightData{} + require.NoError(t, state.Save(newViewRecord, true)) + + require.NoError(t, writeAheadLog.Close()) + + // Nothing before the new view is needed: recovery reads the last record alone. + writeAheadLog, err = wal.Open(log, testDir, options) + require.NoError(t, err) + entries, err := writeAheadLog.ReadAll() + require.NoError(t, err) + require.Len(t, entries, 1) + entry := &protos.SavedMessage{} + require.NoError(t, proto.Unmarshal(entries[0], entry)) + require.NotNil(t, entry.GetNewView()) + filesAfterFirstRestart := logFileCount(t, testDir) + require.NoError(t, writeAheadLog.Close()) + + // A second restart must not add files to the WAL. + writeAheadLog, err = wal.Open(log, testDir, options) + require.NoError(t, err) + require.NoError(t, writeAheadLog.Close()) + + require.LessOrEqual(t, logFileCount(t, testDir), filesAfterFirstRestart) +} diff --git a/internal/bft/viewchanger.go b/internal/bft/viewchanger.go index 4df0cbc7..cc550ae4 100644 --- a/internal/bft/viewchanger.go +++ b/internal/bft/viewchanger.go @@ -23,7 +23,37 @@ import ( // ViewController controls the view type ViewController interface { ViewChanged(newViewNumber uint64, newProposalSequence uint64) - AbortView(view uint64) + // AbortView aborts the given view and does not return until the view has + // stopped. It returns false when the view is already older than the current + // one or when the controller is stopping. A controller may also implement + // viewHolder, which additionally keeps the view stopped for the caller. + AbortView(view uint64) bool +} + +// viewHolder is implemented by a ViewController that can abort a view, keep +// it from being restarted, and keep other views from writing until the view +// changer has written the record that must follow the abort. +type viewHolder interface { + AbortViewAndHold(view uint64) bool + ReleaseViewWrite() +} + +// abortViewAndHold stops the view and holds it when the controller supports +// holding, or just stops it otherwise. +func (v *ViewChanger) abortViewAndHold(view uint64) bool { + if holder, ok := v.Controller.(viewHolder); ok { + return holder.AbortViewAndHold(view) + } + + return v.Controller.AbortView(view) +} + +// releaseViewWrite lets the controller start views again when it supports +// holding. +func (v *ViewChanger) releaseViewWrite() { + if holder, ok := v.Controller.(viewHolder); ok { + holder.ReleaseViewWrite() + } } // Pruner prunes revoked requests @@ -435,6 +465,26 @@ func (v *ViewChanger) processViewChangeMsg(restore bool) { v.Logger.Debugf("Node %d is joining view change, last view is %d", v.SelfID, v.currView) v.startViewChange(&change{v.currView, true}) } + // Stop the current view before writing the view change record. The view may still + // append a commit after that record, which would leave the commit without its + // pre-prepare in the WAL. AbortViewAndHold waits for the view to stop, keeps the + // sync path from restarting it, and keeps other views from writing until + // releaseViewWrite below. If the controller has already moved to a newer view, + // this view change is stale. + aborted := v.abortViewAndHold(v.currView) + if !aborted { + if !restore { + return + } + + // A restore can race with syncOnStart, which advances the controller to a + // newer view and informs the view changer through informChan. The restore + // was already started by the code above, so it must still send its view + // data instead of silently leaving the view change half done. The pending + // informChan entry will move the view changer to the newer view next. + v.Logger.Debugf("Node %d is restoring view change %d while the controller is already in a newer view", v.SelfID, v.currView) + } + if !restore { msgToSave := &protos.SavedMessage{ Content: &protos.SavedMessage_ViewChange{ @@ -443,11 +493,16 @@ func (v *ViewChanger) processViewChangeMsg(restore bool) { }, }, } - if err := v.State.Save(msgToSave); err != nil { + // The flag is passed explicitly, after the view has stopped and no other + // view may write: the proposal in flight, if there is one, travels in the + // view data message and must stay recoverable. + if err := v.State.Save(msgToSave, !v.hasInFlightProposal()); err != nil { v.Logger.Panicf("Failed to save message to state, error: %v", err) } } - v.Controller.AbortView(v.currView) // before preparing the view data message abort the current view + if aborted { + v.releaseViewWrite() + } v.currView = v.nextView v.MetricsViewChange.CurrentView.Set(float64(v.currView)) v.viewChangeMsgs.clear(v.N) @@ -464,6 +519,13 @@ func (v *ViewChanger) processViewChangeMsg(restore bool) { v.Logger.Debugf("Node %d sent view data msg, with next view %d, to the new leader %d", v.SelfID, v.currView, leader) } +// hasInFlightProposal tells whether the node still has a proposal that recovery needs. +// A nil InFlight is possible in tests and in a hand-built view changer, so it is not +// dereferenced blindly. +func (v *ViewChanger) hasInFlightProposal() bool { + return v.InFlight != nil && v.InFlight.InFlightProposal() != nil +} + func (v *ViewChanger) prepareViewDataMsg() *protos.Message { lastDecision, lastDecisionSignatures := v.Checkpoint.Get() inFlightProposal, prepared := v.InFlight.InFlightProposalAndPrepared() @@ -1201,6 +1263,25 @@ func (v *ViewChanger) processNewViewMsg(msg *protos.NewView) { return } + // The record of the new view may truncate the WAL, so the current view has to be stopped + // first. Otherwise it can still append a commit after that record and leave a commit + // without its pre-prepare in the WAL. AbortViewAndHold waits for the view to stop, keeps + // the sync path from restarting it, and keeps other views from writing until + // releaseViewWrite below. If the controller has already moved to a newer view, this new + // view message is stale. + if !v.abortViewAndHold(v.currView) { + return + } + + // Under condition A the quorum either committed the agreed proposal or declared it + // irrelevant, so this node's own in-flight proposal cannot be prepared any more and is + // dropped. Under condition B the quorum excluded this node and reported no prepared + // proposal: this node's own prepared proposal is not covered by that quorum and must be + // kept for the next view change. + if !noInFlight && v.InFlight != nil { + v.InFlight.clear() + } + mySequence, _ := v.extractCurrentSequence() newViewToSave := &protos.SavedMessage{ @@ -1211,9 +1292,12 @@ func (v *ViewChanger) processNewViewMsg(msg *protos.NewView) { }, }, } - if err = v.State.Save(newViewToSave); err != nil { + // The truncation flag is decided now, while no view may write: the record is a + // truncation point exactly when no proposal is left in flight for recovery. + if err = v.State.Save(newViewToSave, !v.hasInFlightProposal()); err != nil { v.Logger.Panicf("Failed to save message to state, error: %v", err) } + v.releaseViewWrite() select { // if there was a delivery or sync with a reconfig when committing the in-flight proposal we should stop case <-v.stopChan: diff --git a/internal/bft/viewchanger_test.go b/internal/bft/viewchanger_test.go index 1b30730d..01da7cf5 100644 --- a/internal/bft/viewchanger_test.go +++ b/internal/bft/viewchanger_test.go @@ -7,6 +7,7 @@ package bft_test import ( "errors" + "reflect" "slices" "strings" "sync" @@ -20,6 +21,7 @@ import ( "github.com/hyperledger/SmartBFT/pkg/metrics" "github.com/hyperledger/SmartBFT/pkg/metrics/disabled" "github.com/hyperledger/SmartBFT/pkg/types" + "github.com/hyperledger/SmartBFT/pkg/wal" protos "github.com/hyperledger/SmartBFT/smartbftprotos" "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" @@ -101,7 +103,7 @@ func TestStartViewChange(t *testing.T) { require.NoError(t, err) log := basicLog.Sugar() controller := &mocks.ViewController{} - controller.On("AbortView", mock.Anything) + controller.On("AbortView", mock.Anything).Return(true) vc := &bft.ViewChanger{ N: 4, @@ -162,9 +164,9 @@ func TestViewChangeProcess(t *testing.T) { reqTimer := &mocks.RequestsTimer{} reqTimer.On("StopTimers") controller := &mocks.ViewController{} - controller.On("AbortView", mock.Anything) + controller.On("AbortView", mock.Anything).Return(true) state := &mocks.State{} - state.On("Save", mock.Anything).Return(nil) + state.On("Save", mock.Anything, mock.Anything).Return(nil) vc := &bft.ViewChanger{ SelfID: 0, @@ -262,6 +264,7 @@ func TestViewDataProcess(t *testing.T) { num = args.Get(1).(uint64) seqNumChan <- num }).Return(nil).Once() + controller.On("AbortView", mock.Anything).Return(true) signer := &mocks.SignerMock{} signer.On("Sign", mock.Anything).Return([]byte{1, 2, 3}) checkpoint := types.Checkpoint{} @@ -269,7 +272,7 @@ func TestViewDataProcess(t *testing.T) { reqTimer := &mocks.RequestsTimer{} reqTimer.On("RestartTimers").Once() state := &mocks.State{} - state.On("Save", mock.Anything).Return(nil) + state.On("Save", mock.Anything, mock.Anything).Return(nil) vc := &bft.ViewChanger{ SelfID: 1, @@ -316,7 +319,7 @@ func TestViewDataProcess(t *testing.T) { vc.Stop() reqTimer.AssertCalled(t, "RestartTimers") - state.AssertCalled(t, "Save", mock.Anything) + state.AssertCalled(t, "Save", mock.Anything, mock.Anything) } func TestNewViewProcess(t *testing.T) { @@ -339,12 +342,13 @@ func TestNewViewProcess(t *testing.T) { num = args.Get(1).(uint64) seqNumChan <- num }).Return(nil).Once() + controller.On("AbortView", mock.Anything).Return(true) checkpoint := types.Checkpoint{} checkpoint.Set(lastDecision, lastDecisionSignatures) reqTimer := &mocks.RequestsTimer{} reqTimer.On("RestartTimers").Once() state := &mocks.State{} - state.On("Save", mock.Anything).Return(nil) + state.On("Save", mock.Anything, mock.Anything).Return(nil) vc := &bft.ViewChanger{ SelfID: 0, @@ -398,7 +402,7 @@ func TestNewViewProcess(t *testing.T) { vc.Stop() reqTimer.AssertCalled(t, "RestartTimers") - state.AssertCalled(t, "Save", mock.Anything) + state.AssertCalled(t, "Save", mock.Anything, mock.Anything) } func TestNormalProcess(t *testing.T) { @@ -427,14 +431,14 @@ func TestNormalProcess(t *testing.T) { num = args.Get(1).(uint64) seqNumChan <- num }).Return(nil).Once() - controller.On("AbortView", mock.Anything) + controller.On("AbortView", mock.Anything).Return(true) reqTimer := &mocks.RequestsTimer{} reqTimer.On("StopTimers") reqTimer.On("RestartTimers") checkpoint := types.Checkpoint{} checkpoint.Set(lastDecision, lastDecisionSignatures) state := &mocks.State{} - state.On("Save", mock.Anything).Return(nil) + state.On("Save", mock.Anything, mock.Anything).Return(nil) vc := &bft.ViewChanger{ SelfID: 1, @@ -472,7 +476,7 @@ func TestNormalProcess(t *testing.T) { num = <-seqNumChan require.Equal(t, uint64(2), num) - controller.AssertNumberOfCalls(t, "AbortView", 2) + controller.AssertNumberOfCalls(t, "AbortView", 3) state.AssertNumberOfCalls(t, "Save", 2) vc.Stop() @@ -966,7 +970,7 @@ func TestResendViewChangeMessage(t *testing.T) { require.NoError(t, err) log := basicLog.Sugar() controller := &mocks.ViewController{} - controller.On("AbortView", mock.Anything) + controller.On("AbortView", mock.Anything).Return(true) vc := &bft.ViewChanger{ N: 4, @@ -1029,7 +1033,7 @@ func TestViewChangerTimeout(t *testing.T) { controllerWG := sync.WaitGroup{} controller.On("AbortView", mock.Anything).Run(func(args mock.Arguments) { controllerWG.Done() - }) + }).Return(true) vc := &bft.ViewChanger{ N: 4, @@ -1087,7 +1091,7 @@ func TestBackOff(t *testing.T) { controllerWG := sync.WaitGroup{} controller.On("AbortView", mock.Anything).Run(func(args mock.Arguments) { controllerWG.Done() - }) + }).Return(true) timeout := 10 * time.Second @@ -1158,7 +1162,7 @@ func TestCommitLastDecision(t *testing.T) { num = args.Get(1).(uint64) seqNumChan <- num }).Return(nil).Once() - controller.On("AbortView", mock.Anything) + controller.On("AbortView", mock.Anything).Return(true) reqTimer := &mocks.RequestsTimer{} reqTimer.On("StopTimers") reqTimer.On("RestartTimers") @@ -1172,7 +1176,7 @@ func TestCommitLastDecision(t *testing.T) { }).Return(types.Reconfig{InLatestDecision: false}) state := &mocks.State{} - state.On("Save", mock.Anything).Return(nil) + state.On("Save", mock.Anything, mock.Anything).Return(nil) pruner := &mocks.Pruner{} pruner.On("MaybePruneRevokedRequests") @@ -1223,7 +1227,7 @@ func TestCommitLastDecision(t *testing.T) { num = <-seqNumChan require.Equal(t, uint64(3), num) - controller.AssertNumberOfCalls(t, "AbortView", 2) + controller.AssertNumberOfCalls(t, "AbortView", 3) app.AssertNumberOfCalls(t, "Deliver", 1) pruner.AssertNumberOfCalls(t, "MaybePruneRevokedRequests", 1) state.AssertNumberOfCalls(t, "Save", 2) @@ -1366,11 +1370,11 @@ func TestInFlightProposalInViewData(t *testing.T) { reqTimer := &mocks.RequestsTimer{} reqTimer.On("StopTimers") controller := &mocks.ViewController{} - controller.On("AbortView", mock.Anything) + controller.On("AbortView", mock.Anything).Return(true) checkpoint := types.Checkpoint{} checkpoint.Set(lastDecision, lastDecisionSignatures) state := &mocks.State{} - state.On("Save", mock.Anything).Return(nil) + state.On("Save", mock.Anything, mock.Anything).Return(nil) vc := &bft.ViewChanger{ SelfID: 0, @@ -1586,7 +1590,7 @@ func TestInformViewChanger(t *testing.T) { require.NoError(t, err) log := basicLog.Sugar() controller := &mocks.ViewController{} - controller.On("AbortView", mock.Anything) + controller.On("AbortView", mock.Anything).Return(true) vc := &bft.ViewChanger{ N: 4, @@ -1634,7 +1638,7 @@ func TestInformNewViewDoesNotBlock(t *testing.T) { basicLog, err := zap.NewDevelopment() require.NoError(t, err) controller := &mocks.ViewController{} - controller.On("AbortView", mock.Anything) + controller.On("AbortView", mock.Anything).Return(true) vc := &bft.ViewChanger{ N: 4, @@ -1680,6 +1684,68 @@ func TestInformNewViewDoesNotBlock(t *testing.T) { vc.Stop() } +func TestRestoreViewChangeWithControllerAhead(t *testing.T) { + comm := &mocks.CommMock{} + broadcastChan := make(chan *protos.Message) + comm.On("BroadcastConsensus", mock.Anything).Run(func(args mock.Arguments) { + m := args.Get(0).(*protos.Message) + broadcastChan <- m + }).Once() + sendChan := make(chan *protos.Message) + comm.On("SendConsensus", mock.Anything, mock.Anything).Run(func(args mock.Arguments) { + m := args.Get(1).(*protos.Message) + sendChan <- m + }).Once() + signer := &mocks.SignerMock{} + signer.On("Sign", mock.Anything).Return([]byte{1, 2, 3}) + basicLog, err := zap.NewDevelopment() + require.NoError(t, err) + log := basicLog.Sugar() + reqTimer := &mocks.RequestsTimer{} + reqTimer.On("StopTimers") + controller := &mocks.ViewController{} + // syncOnStart has already moved the controller to a newer view than the one + // whose view change is being restored. + controller.On("AbortView", mock.Anything).Return(false) + + vc := &bft.ViewChanger{ + SelfID: 0, + N: 4, + NodesList: []uint64{0, 1, 2, 3}, + Comm: comm, + Signer: signer, + Logger: log, + RequestsTimer: reqTimer, + Ticker: make(chan time.Time), + InFlight: &bft.InFlightData{}, + Checkpoint: &types.Checkpoint{}, + Controller: controller, + InMsqQSize: 100, + } + + restoreChan := make(chan struct{}, 1) + restoreChan <- struct{}{} + vc.Restore = restoreChan + + vc.Start(5) + + select { + case m := <-broadcastChan: + require.NotNil(t, m.GetViewChange()) + case <-time.After(time.Second): + require.Fail(t, "the restored view change was not broadcast") + } + + select { + case m := <-sendChan: + require.NotNil(t, m.GetViewData(), "the restored view change ended without sending its view data") + case <-time.After(time.Second): + require.Fail(t, "the restored view change ended without sending its view data") + } + + vc.Stop() +} + func TestRestoreViewChange(t *testing.T) { comm := &mocks.CommMock{} broadcastChan := make(chan *protos.Message) @@ -1700,7 +1766,7 @@ func TestRestoreViewChange(t *testing.T) { reqTimer := &mocks.RequestsTimer{} reqTimer.On("StopTimers") controller := &mocks.ViewController{} - controller.On("AbortView", mock.Anything) + controller.On("AbortView", mock.Anything).Return(true) vc := &bft.ViewChanger{ SelfID: 0, @@ -2014,7 +2080,7 @@ func TestCommitInFlight(t *testing.T) { num = args.Get(1).(uint64) seqNumChan <- num }).Return(nil).Once() - controller.On("AbortView", mock.Anything) + controller.On("AbortView", mock.Anything).Return(true) reqTimer := &mocks.RequestsTimer{} reqTimer.On("StopTimers") reqTimer.On("RestartTimers") @@ -2130,7 +2196,7 @@ func TestCommitInFlight(t *testing.T) { num = <-seqNumChan require.Equal(t, uint64(3), num) - controller.AssertNumberOfCalls(t, "AbortView", 2) + controller.AssertNumberOfCalls(t, "AbortView", 3) pruner.AssertNumberOfCalls(t, "MaybePruneRevokedRequests", 1) // the requests of the in flight proposal are removed from the pool once it is decided for _, req := range inFlightRequests { @@ -2216,7 +2282,7 @@ func TestInFlightViewMessageFromItselfWithWrongView(t *testing.T) { verifier.On("RequestsFromProposal", mock.Anything).Return(nil) controller := &mocks.ViewController{} controller.On("ViewChanged", mock.Anything, mock.Anything).Return(nil).Maybe() - controller.On("AbortView", mock.Anything) + controller.On("AbortView", mock.Anything).Return(true) reqTimer := &mocks.RequestsTimer{} reqTimer.On("StopTimers") reqTimer.On("RestartTimers") @@ -2362,7 +2428,7 @@ func TestCommitWhileHavingInFlight(t *testing.T) { verifier.On("RequestsFromProposal", mock.Anything).Return(nil) controller := &mocks.ViewController{} controller.On("ViewChanged", mock.Anything, mock.Anything) - controller.On("AbortView", mock.Anything) + controller.On("AbortView", mock.Anything).Return(true) reqTimer := &mocks.RequestsTimer{} reqTimer.On("StopTimers") reqTimer.On("RestartTimers") @@ -2455,12 +2521,13 @@ func TestDontCommitInFlight(t *testing.T) { num = args.Get(1).(uint64) seqNumChan <- num }).Return(nil).Once() + controller.On("AbortView", mock.Anything).Return(true) reqTimer := &mocks.RequestsTimer{} reqTimer.On("RestartTimers") app := &mocks.ApplicationMock{} app.On("Deliver", mock.Anything, mock.Anything) state := &mocks.State{} - state.On("Save", mock.Anything).Return(nil) + state.On("Save", mock.Anything, mock.Anything).Return(nil) vc := &bft.ViewChanger{ SelfID: 3, @@ -2472,6 +2539,7 @@ func TestDontCommitInFlight(t *testing.T) { Ticker: make(chan time.Time), RequestsTimer: reqTimer, Application: app, + InFlight: &bft.InFlightData{}, State: state, } @@ -2547,12 +2615,13 @@ func newViewChangerWithInFlight(t *testing.T, verifier api.Verifier) (*bft.ViewC controller.On("ViewChanged", mock.Anything, mock.Anything).Run(func(args mock.Arguments) { viewChangedChan <- args.Get(0).(uint64) }).Return(nil).Maybe() + controller.On("AbortView", mock.Anything).Return(true) reqTimer := &mocks.RequestsTimer{} reqTimer.On("RestartTimers") app := &mocks.ApplicationMock{} app.On("Deliver", mock.Anything, mock.Anything) state := &mocks.State{} - state.On("Save", mock.Anything).Return(nil) + state.On("Save", mock.Anything, mock.Anything).Return(nil) vc := &bft.ViewChanger{ SelfID: 3, @@ -2963,6 +3032,7 @@ func TestNewViewWithoutLeaderViewData(t *testing.T) { viewNumChan <- args.Get(0).(uint64) seqNumChan <- args.Get(1).(uint64) }) + controller.On("AbortView", mock.Anything).Return(true) reqTimer := &mocks.RequestsTimer{} reqTimer.On("StopTimers") reqTimer.On("RestartTimers") @@ -2971,7 +3041,7 @@ func TestNewViewWithoutLeaderViewData(t *testing.T) { pruner := &mocks.Pruner{} pruner.On("MaybePruneRevokedRequests") state := &mocks.State{} - state.On("Save", mock.Anything).Return(nil) + state.On("Save", mock.Anything, mock.Anything).Return(nil) checkpoint := types.Checkpoint{} checkpoint.Set(lastDecision, lastDecisionSignatures) @@ -3108,7 +3178,7 @@ func TestViewChangerKeepsRunningAfterInFlightWaiterLeft(t *testing.T) { verifier.On("RequestsFromProposal", mock.Anything).Return(nil) controller := &mocks.ViewController{} controller.On("ViewChanged", mock.Anything, mock.Anything).Maybe() - controller.On("AbortView", mock.Anything) + controller.On("AbortView", mock.Anything).Return(true) reqTimer := &mocks.RequestsTimer{} reqTimer.On("StopTimers") reqTimer.On("RestartTimers") @@ -3233,3 +3303,422 @@ func waitForMessage(t *testing.T, msgChan <-chan *protos.Message, match func(*pr } } } + +// walState is a real PersistedState writing into a real write-ahead log, wrapped so that a test +// can wait until a message actually reached the log instead of guessing when that happened. +type walState struct { + *bft.PersistedState + file *wal.WriteAheadLogFile + written chan struct{} +} + +func (s *walState) Save(msg *protos.SavedMessage, truncateTo bool) error { + if err := s.PersistedState.Save(msg, truncateTo); err != nil { + return err + } + s.written <- struct{}{} + return nil +} + +// entriesAfterRestart closes the log and reads it again the way a restarted node does: a node +// recovers only what was written at or after the last truncating record. +func (s *walState) entriesAfterRestart(t *testing.T, dir string) [][]byte { + t.Helper() + require.NoError(t, s.file.Close()) + reopened, err := wal.Open(s.Logger, dir, nil) + require.NoError(t, err) + defer func() { + require.NoError(t, reopened.Close()) + }() + entries, err := reopened.ReadAll() + require.NoError(t, err) + return entries +} + +func newWALState(t *testing.T, log api.Logger, dir string, inFlight *bft.InFlightData) *walState { + t.Helper() + file, err := wal.Create(log, dir, nil) + require.NoError(t, err) + return &walState{ + PersistedState: &bft.PersistedState{InFlightProposal: inFlight, Logger: log, WAL: file}, + file: file, + written: make(chan struct{}, 8), + } +} + +// inFlightProposal returns the proposal a node holds in flight for the given sequence. +func inFlightProposal(seq uint64) types.Proposal { + return types.Proposal{ + Header: []byte{1}, + Payload: []byte{2}, + Metadata: bft.MarshalOrPanic(&protos.ViewMetadata{ViewId: 1, LatestSequence: seq}), + VerificationSequence: 1, + } +} + +// prePrepareRecord is what a node writes once it accepts a proposal, before it has enough +// prepares to commit it. Storing it is what puts the proposal in flight. +func prePrepareRecord(seq uint64) *protos.SavedMessage { + proposal := inFlightProposal(seq) + return &protos.SavedMessage{ + Content: &protos.SavedMessage_ProposedRecord{ + ProposedRecord: &protos.ProposedRecord{ + PrePrepare: &protos.PrePrepare{ + View: 1, + Seq: seq, + Proposal: &protos.Proposal{ + Header: proposal.Header, + Payload: proposal.Payload, + Metadata: proposal.Metadata, + VerificationSequence: uint64(proposal.VerificationSequence), + }, + }, + Prepare: &protos.Prepare{View: 1, Seq: seq}, + }, + }, + } +} + +// commitRecord is what a node writes once it has enough prepares to commit the proposal in flight. +func commitRecord(seq uint64) *protos.SavedMessage { + return &protos.SavedMessage{ + Content: &protos.SavedMessage_Commit{ + Commit: &protos.Message{ + Content: &protos.Message_Commit{ + Commit: &protos.Commit{View: 1, Seq: seq}, + }, + }, + }, + } +} + +// newViewData returns view data that reports the last decision the package level fixtures use. +func newViewData(view uint64) *protos.ViewData { + viewData := proto.Clone(vd).(*protos.ViewData) + viewData.NextView = view + return viewData +} + +// viewDataWithInFlight returns view data claiming a prepared proposal with sequence 2 in flight. +// Nodes that report different payloads are reporting different proposals. +func viewDataWithInFlight(view uint64, payload byte) *protos.ViewData { + viewData := newViewData(view) + viewData.InFlightProposal = &protos.Proposal{ + Header: []byte{2}, + Payload: []byte{payload}, + Metadata: bft.MarshalOrPanic(&protos.ViewMetadata{ViewId: 1, LatestSequence: 2}), + VerificationSequence: 1, + } + viewData.InFlightPrepared = true + return viewData +} + +// newViewMessage builds a new view message out of the given view data, signed by the nodes 0 to +// len(viewData)-1. The leader of the new view must be among them. +func newViewMessage(viewData ...*protos.ViewData) *protos.Message { + signed := make([]*protos.SignedViewData, 0, len(viewData)) + for i, data := range viewData { + signed = append(signed, &protos.SignedViewData{RawViewData: bft.MarshalOrPanic(data), Signer: uint64(i)}) + } + return &protos.Message{ + Content: &protos.Message_NewView{ + NewView: &protos.NewView{SignedViewData: signed}, + }, + } +} + +func unmarshalSavedMessage(t *testing.T, entry []byte) *protos.SavedMessage { + t.Helper() + msg := &protos.SavedMessage{} + require.NoError(t, proto.Unmarshal(entry, msg)) + return msg +} + +// A node that receives a new view under condition B (a quorum reports no prepared +// proposal) must keep its own prepared proposal: the quorum excluded this node, so +// the proposal is not covered by it. Consequently the new view record cannot be a +// truncation point. +func TestNewViewKeepsLocalInFlightUnderConditionB(t *testing.T) { + basicLog, err := zap.NewDevelopment() + require.NoError(t, err) + log := basicLog.Sugar() + + verifier := &mocks.VerifierMock{} + verifier.On("VerifySignature", mock.Anything).Return(nil) + controller := &mocks.ViewController{} + controller.On("AbortView", mock.Anything).Return(true) + viewChanged := make(chan struct{}, 1) + controller.On("ViewChanged", mock.Anything, mock.Anything).Run(func(mock.Arguments) { + viewChanged <- struct{}{} + }).Return(nil) + reqTimer := &mocks.RequestsTimer{} + reqTimer.On("RestartTimers").Return(nil) + + dir := t.TempDir() + inFlight := &bft.InFlightData{} + state := newWALState(t, log, dir, inFlight) + + checkpoint := types.Checkpoint{} + checkpoint.Set(lastDecision, lastDecisionSignatures) + + vc := &bft.ViewChanger{ + SelfID: 0, + N: 4, + NodesList: []uint64{0, 1, 2, 3}, + Logger: log, + Verifier: verifier, + Controller: controller, + Ticker: make(chan time.Time), + RequestsTimer: reqTimer, + Checkpoint: &checkpoint, + InFlight: inFlight, + State: state, + InMsqQSize: 100, + } + vc.Start(2) + + // the node holds a proposal in flight: a pre-prepare and the commit that followed it + require.NoError(t, state.Save(prePrepareRecord(2), true)) + require.NoError(t, state.Save(commitRecord(2), false)) + expected := inFlightProposal(2) + require.Equal(t, &expected, inFlight.InFlightProposal()) + + // the leader of view 2 reports a quorum of nodes that have no proposal in flight, so the + // proposal of this node is either committed by them or is stale + vc.HandleMessage(2, newViewMessage(newViewData(2), newViewData(2), newViewData(2))) + + <-viewChanged + vc.Stop() + + require.Equal(t, &expected, inFlight.InFlightProposal(), "condition B must keep this node's prepared proposal") + + entries := state.entriesAfterRestart(t, dir) + require.Len(t, entries, 3, "the new view record truncated the proposal this node still carries") + require.NotNil(t, unmarshalSavedMessage(t, entries[0]).GetProposedRecord()) + require.NotNil(t, unmarshalSavedMessage(t, entries[1]).GetCommit()) + require.NotNil(t, unmarshalSavedMessage(t, entries[2]).GetNewView()) +} + +// A node that moved to a new view and has no proposal in flight has nothing left +// to recover, so its new view record is a truncation point. +func TestNewViewWithoutInFlightTruncatesWal(t *testing.T) { + basicLog, err := zap.NewDevelopment() + require.NoError(t, err) + log := basicLog.Sugar() + + verifier := &mocks.VerifierMock{} + verifier.On("VerifySignature", mock.Anything).Return(nil) + controller := &mocks.ViewController{} + controller.On("AbortView", mock.Anything).Return(true) + viewChanged := make(chan struct{}, 1) + controller.On("ViewChanged", mock.Anything, mock.Anything).Run(func(mock.Arguments) { + viewChanged <- struct{}{} + }).Return(nil) + reqTimer := &mocks.RequestsTimer{} + reqTimer.On("RestartTimers").Return(nil) + + dir := t.TempDir() + inFlight := &bft.InFlightData{} + state := newWALState(t, log, dir, inFlight) + + checkpoint := types.Checkpoint{} + checkpoint.Set(lastDecision, lastDecisionSignatures) + + vc := &bft.ViewChanger{ + SelfID: 0, + N: 4, + NodesList: []uint64{0, 1, 2, 3}, + Logger: log, + Verifier: verifier, + Controller: controller, + Ticker: make(chan time.Time), + RequestsTimer: reqTimer, + Checkpoint: &checkpoint, + InFlight: inFlight, + State: state, + InMsqQSize: 100, + } + vc.Start(2) + + require.NoError(t, state.file.Append([]byte("record before the new view"), false)) + + vc.HandleMessage(2, newViewMessage(newViewData(2), newViewData(2), newViewData(2))) + + <-viewChanged + vc.Stop() + + require.Nil(t, inFlight.InFlightProposal()) + + entries := state.entriesAfterRestart(t, dir) + require.Len(t, entries, 1, "the new view record was not written as a truncation point") + newView := unmarshalSavedMessage(t, entries[0]).GetNewView() + require.NotNil(t, newView) + require.Equal(t, uint64(2), newView.GetViewId()) +} + +// A quorum that neither committed nor invalidated the proposal in flight does not move this node +// to the new view, hence nothing about the proposal in flight may be cleared. +func TestNewViewWithoutQuorumOnInFlightKeepsInFlight(t *testing.T) { + basicLog, err := zap.NewDevelopment() + require.NoError(t, err) + log := basicLog.Sugar() + + verifier := &mocks.VerifierMock{} + verified := make(chan struct{}, 8) + verifier.On("VerifySignature", mock.Anything).Run(func(mock.Arguments) { + verified <- struct{}{} + }).Return(nil) + controller := &mocks.ViewController{} + controller.On("AbortView", mock.Anything).Return(true) + controller.On("ViewChanged", mock.Anything, mock.Anything).Return(nil) + reqTimer := &mocks.RequestsTimer{} + reqTimer.On("RestartTimers").Return(nil) + + dir := t.TempDir() + inFlight := &bft.InFlightData{} + state := newWALState(t, log, dir, inFlight) + + checkpoint := types.Checkpoint{} + checkpoint.Set(lastDecision, lastDecisionSignatures) + + vc := &bft.ViewChanger{ + SelfID: 0, + N: 4, + NodesList: []uint64{0, 1, 2, 3}, + Logger: log, + Verifier: verifier, + Controller: controller, + Ticker: make(chan time.Time), + RequestsTimer: reqTimer, + Checkpoint: &checkpoint, + InFlight: inFlight, + State: state, + InMsqQSize: 100, + } + vc.Start(2) + + proposal := inFlightProposal(2) + require.NoError(t, state.Save(prePrepareRecord(2), true)) + require.NoError(t, state.Save(commitRecord(2), false)) + + // only the records written while handling the message are of interest here + for range 2 { + <-state.written + } + + // three nodes report three different prepared proposals, so no proposal is agreed upon + vc.HandleMessage(2, newViewMessage( + viewDataWithInFlight(2, 1), + viewDataWithInFlight(2, 2), + viewDataWithInFlight(2, 3), + )) + + for range 3 { + <-verified + } + vc.Stop() + + require.Equal(t, &proposal, inFlight.InFlightProposal(), "the proposal in flight was cleared without a quorum") + controller.AssertNotCalled(t, "ViewChanged") + require.Empty(t, state.written, "a record was written although the view did not change") + + entries := state.entriesAfterRestart(t, dir) + require.Len(t, entries, 2, "the records of the view this node stayed in were truncated") + require.NotNil(t, unmarshalSavedMessage(t, entries[0]).GetProposedRecord()) + require.NotNil(t, unmarshalSavedMessage(t, entries[1]).GetCommit()) +} + +// A node that leaves for a new view is not synchronized yet, and its proposal in flight travels +// to the new leader within the view data message, so it must be kept. A node without a proposal +// in flight has nothing to recover, hence its view change record is a truncation point. +func TestViewChangeKeepsInFlightAndTruncatesWithoutIt(t *testing.T) { + for _, testCase := range []struct { + description string + sequence uint64 + }{ + {description: "with a proposal in flight", sequence: 2}, + {description: "without a proposal in flight", sequence: 0}, + } { + t.Run(testCase.description, func(t *testing.T) { + basicLog, err := zap.NewDevelopment() + require.NoError(t, err) + log := basicLog.Sugar() + + comm := &mocks.CommMock{} + comm.On("BroadcastConsensus", mock.Anything).Return(nil) + viewDataSent := make(chan struct{}, 1) + comm.On("SendConsensus", mock.Anything, mock.Anything).Run(func(mock.Arguments) { + viewDataSent <- struct{}{} + }).Return(nil) + signer := &mocks.SignerMock{} + signer.On("Sign", mock.Anything).Return([]byte{1, 2, 3}) + controller := &mocks.ViewController{} + controller.On("AbortView", mock.Anything).Return(true) + reqTimer := &mocks.RequestsTimer{} + reqTimer.On("StopTimers").Return(nil) + + dir := t.TempDir() + inFlight := &bft.InFlightData{} + state := newWALState(t, log, dir, inFlight) + var proposal types.Proposal + + checkpoint := types.Checkpoint{} + checkpoint.Set(lastDecision, lastDecisionSignatures) + + vc := &bft.ViewChanger{ + SelfID: 0, + N: 4, + NodesList: []uint64{0, 1, 2, 3}, + Logger: log, + Comm: comm, + Signer: signer, + Controller: controller, + Ticker: make(chan time.Time), + RequestsTimer: reqTimer, + Checkpoint: &checkpoint, + InFlight: inFlight, + State: state, + InMsqQSize: 100, + } + vc.Start(0) + + if testCase.sequence != 0 { + proposal = inFlightProposal(testCase.sequence) + require.NoError(t, state.Save(prePrepareRecord(testCase.sequence), true)) + require.NotNil(t, inFlight.InFlightProposal()) + } + + // a quorum of view change messages makes this node join the view change + vc.HandleMessage(1, viewChangeMsg) + vc.HandleMessage(2, viewChangeMsg) + + <-viewDataSent + vc.Stop() + + entries := state.entriesAfterRestart(t, dir) + require.NotEmpty(t, entries) + require.NotNil(t, unmarshalSavedMessage(t, entries[len(entries)-1]).GetViewChange()) + + if testCase.sequence == 0 { + require.Len(t, entries, 1, "the view change record was not written as a truncation point") + require.Nil(t, inFlight.InFlightProposal()) + return + } + + require.Len(t, entries, 2, "the proposal in flight was truncated away by the view change") + require.NotNil(t, unmarshalSavedMessage(t, entries[0]).GetProposedRecord()) + require.Equal(t, &proposal, inFlight.InFlightProposal(), "the proposal in flight was cleared by the view change") + }) + } +} + +// The view changer reports what it did through this interface, and the decision to clear the +// proposal in flight is a detail of the view change protocol, hence no method was added to it. +func TestViewControllerIsNotExtended(t *testing.T) { + controllerType := reflect.TypeFor[bft.ViewController]() + methods := make([]string, 0, controllerType.NumMethod()) + for method := range controllerType.Methods() { + methods = append(methods, method.Name) + } + require.Equal(t, []string{"AbortView", "ViewChanged"}, methods) +} diff --git a/test/test_app.go b/test/test_app.go index 1271f46f..faab76b2 100644 --- a/test/test_app.go +++ b/test/test_app.go @@ -61,6 +61,7 @@ type App struct { secondClock *time.Ticker logger *zap.SugaredLogger metricsProvider metrics.Provider + walOptions *wal.Options lastRecord lastRecord verificationSeq atomic.Uint64 clientsToRevoke []string // clients whose requests become invalid once a reconfig is delivered @@ -444,7 +445,27 @@ type AppRecord struct { Metadata []byte } -func newNode(id uint64, network *Network, testName string, testDir string, rotateLeader bool, decisionsPerLeader uint64) *App { +// appOption configures a test application created by newNode. +type appOption func(*App) + +// WithWalOptions sets the WAL options of the test application. +func WithWalOptions(options *wal.Options) appOption { + return func(app *App) { + if options == nil { + return + } + + app.walOptions.FileSizeBytes = options.FileSizeBytes + app.walOptions.BufferSizeBytes = options.BufferSizeBytes + if options.Metrics != nil { + app.walOptions.Metrics = options.Metrics + } + } +} + +// newNode creates a test application. Options are applied before the WAL is +// initialized, so WithWalOptions can change how the WAL rotates its files. +func newNode(id uint64, network *Network, testName string, testDir string, rotateLeader bool, decisionsPerLeader uint64, options ...appOption) *App { logConfig := zap.NewDevelopmentConfig() logger, _ := logConfig.Build() logger = logger.With(zap.String("t", testName)).With(zap.Int64("id", int64(id))) @@ -462,6 +483,11 @@ func newNode(id uint64, network *Network, testName string, testDir string, rotat metricsProvider: &disabled.Provider{}, } + app.walOptions = &wal.Options{Metrics: wal.NewMetrics(app.metricsProvider)} + for _, option := range options { + option(app) + } + config := fastConfig config.SelfID = id config.SyncOnStart = true @@ -472,7 +498,7 @@ func newNode(id uint64, network *Network, testName string, testDir string, rotat writeAheadLog, walInitialEntries, err := wal.InitializeAndReadAll( app.logger, filepath.Join(testDir, fmt.Sprintf("node%d", id)), - &wal.Options{Metrics: wal.NewMetrics(app.metricsProvider)}, + app.walOptions, ) if err != nil { sugaredLogger.Panicf("Failed to initialize WAL: %s", err) diff --git a/test/wal_truncation_test.go b/test/wal_truncation_test.go index 07bf91d1..bd8ce1e4 100644 --- a/test/wal_truncation_test.go +++ b/test/wal_truncation_test.go @@ -276,3 +276,112 @@ func TestRestartKeepsViewAndInFlightProposal(t *testing.T) { require.Equal(t, baseline, walLogFileCount(t, walDir), "ten consecutive restarts must not grow the WAL") } + +// TestNodeWithoutPrePrepareDoesNotGrowWAL covers the live-node variant of the +// WAL growth: the node never restarts, keeps receiving every consensus message +// except pre-prepares, and still goes through view changes. On main, its +// ViewChange and NewView records are not truncation points, so the WAL grows at +// every file rotation. +func TestNodeWithoutPrePrepareDoesNotGrowWAL(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++ { + var options []appOption + if i == numberOfNodes { + // A tiny file makes every view-change record rotate the WAL, so the + // test does not have to write megabytes to observe the growth. + options = append(options, WithWalOptions(&wal.Options{FileSizeBytes: 32})) + } + nodes = append(nodes, newNode(uint64(i), network, t.Name(), testDir, false, 0, options...)) + } + + isolated := nodes[numberOfNodes-1] + isolated.LoseMessages(func(m *protos.Message) bool { + return m.GetPrePrepare() != nil + }) + + startedViews := make(chan uint64, 16) + processedProposal := make(chan struct{}, 1) + baseLogger := isolated.Consensus.Logger.(*zap.SugaredLogger).Desugar() + isolated.Consensus.Logger = baseLogger.WithOptions(zap.Hooks(func(entry zapcore.Entry) error { + if strings.Contains(entry.Message, "Processed proposal with seq") { + select { + case processedProposal <- struct{}{}: + default: + } + } + var view, sequence, decisions uint64 + if _, err := fmt.Sscanf(entry.Message, "Starting view with number %d, sequence %d, and decisions %d", &view, &sequence, &decisions); err == nil { + select { + case startedViews <- view: + default: + } + } + return nil + })).Sugar() + + startNodes(nodes, network) + + // The cluster keeps deciding while node 4 is alive but cannot process a + // pre-prepare itself. Node 4 is allowed to be behind; it catches up via sync. + nodes[0].Submit(Request{ID: "1", ClientID: "alice"}) + for _, n := range nodes[:numberOfNodes-1] { + <-n.Delivered + } + + walDir := filepath.Join(testDir, fmt.Sprintf("node%d", isolated.ID)) + + // The first view change moves the WAL to its steady state: the truncating + // record keeps the file it was written to and the switch creates the next + // one. The baseline is taken after it. Complaining on every node makes the + // view change deterministic instead of waiting for the heartbeat timeout. + for _, n := range nodes { + n.Consensus.Complain(0, true) + } + waitForStartedView(t, startedViews, 1) + baseline := walLogFileCount(t, walDir) + + // View 2 is led by node 3, so node 4 stays a follower and keeps losing + // pre-prepares. Its WAL must not add a file for this view change. + for _, n := range nodes { + n.Consensus.Complain(1, true) + } + waitForStartedView(t, startedViews, 2) + + require.LessOrEqual(t, walLogFileCount(t, walDir), baseline, + "a live node that misses pre-prepares must not grow its WAL on view changes") + + select { + case <-processedProposal: + t.Fatal("the isolated node accepted a pre-prepare") + default: + } + + lastRecord := lastWalLogRecord(t, isolated.logger, walDir) + lastMessage := &protos.SavedMessage{} + require.NoError(t, proto.Unmarshal(lastRecord.GetData(), lastMessage)) + require.NotNil(t, lastMessage.GetNewView(), "the node must have persisted the new view") +} + +func waitForStartedView(t *testing.T, views <-chan uint64, target uint64) { + t.Helper() + + timeout := time.After(30 * time.Second) + for { + select { + case view := <-views: + if view >= target { + return + } + case <-timeout: + t.Fatalf("timed out waiting for view %d", target) + } + } +}