From c748f653773fe1283d6e9705efc743eafd24c64a 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 | 118 ++++-- internal/bft/controller_sync_test.go | 407 +++++++++++++++++- internal/bft/controller_test.go | 100 ++++- internal/bft/controller_view_guard_test.go | 107 +++++ internal/bft/view.go | 2 +- internal/bft/view_test.go | 114 ++++++ internal/bft/viewchanger.go | 41 +- internal/bft/viewchanger_test.go | 454 ++++++++++++++++++++- test/test_app.go | 30 +- test/wal_truncation_test.go | 109 +++++ 10 files changed, 1418 insertions(+), 64 deletions(-) create mode 100644 internal/bft/controller_view_guard_test.go diff --git a/internal/bft/controller.go b/internal/bft/controller.go index 886c8452..f7072f2d 100644 --- a/internal/bft/controller.go +++ b/internal/bft/controller.go @@ -109,16 +109,19 @@ 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 // viewTransitionLock serializes aborting a view with starting or stopping // one. A view changer can hold it from AbortViewAndHold until // ReleaseViewWrite, so no view appends a record while the view changer - // writes its own. AbortView must not return while changeView is between - // aborting the old view and starting the new one: the abort would hit the - // already stopped old view and the new view would keep running. The lock - // also keeps the view number and the view consistent, since changeView - // sets them in separate steps. + // writes the record that may truncate the WAL. It also keeps the view + // number and the view consistent, since changeView sets them in separate + // steps. viewTransitionLock sync.Mutex + viewWriteHeld bool currDecisionsInViewLock sync.RWMutex currDecisionsInView uint64 @@ -191,6 +194,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() @@ -403,6 +432,11 @@ func (c *Controller) changeView(newViewNumber uint64, newProposalSequence uint64 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() @@ -472,32 +506,39 @@ func (c *Controller) Sync() { c.grabSyncToken() } -// AbortView aborts the current view, unless the given view is older than it, -// 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 new view record. It does not wait for the controller loop, which -// may be busy with a long sync or an application delivery, but it does wait -// for a changeView that already started. +// 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 does not wait for the controller loop, which may be busy +// with a long sync or an application delivery. // // If cancel is closed before the view stops, AbortView stops waiting for it. // The view changer passes its stop channel, so that stopping the view changer // does not hang on a view that waits for a lock held by a reconfiguration. func (c *Controller) AbortView(view uint64, cancel <-chan struct{}) { - if !c.AbortViewAndHold(view, cancel) { + c.Logger.Debugf("AbortView, the current view num is %d", c.getCurrentViewNumber()) + + c.Batcher.Close() + + c.viewTransitionLock.Lock() + defer c.viewTransitionLock.Unlock() + + if c.stopped() { return } - c.ReleaseViewWrite() + c.abortView(view, cancel) } -// AbortViewAndHold aborts the current view like AbortView and, on success, -// keeps viewTransitionLock held, so no other view can be started or stopped -// until ReleaseViewWrite. The sync path restarts views, so without the lock it +// 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. // -// The caller must call ReleaseViewWrite after writing the record that had to -// follow the abort. +// On success the controller keeps viewTransitionLock 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, cancel <-chan struct{}) bool { c.Logger.Debugf("AbortViewAndHold, the current view num is %d", c.getCurrentViewNumber()) @@ -517,14 +558,30 @@ func (c *Controller) AbortViewAndHold(view uint64, cancel <-chan struct{}) bool 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. The -// caller must have called AbortViewAndHold successfully. +// 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.viewTransitionLock.Unlock() + c.currViewLock.Lock() + held := c.viewWriteHeld + c.viewWriteHeld = false + if held { + c.minRestartView = 0 + } + c.currViewLock.Unlock() + + if held { + c.viewTransitionLock.Unlock() + } } // ViewChanged makes the controller abort the current view and start a new one with the given numbers @@ -565,6 +622,9 @@ 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 <-c.stopChan: return @@ -740,10 +800,10 @@ func (c *Controller) sync() (viewNum uint64, seq uint64, decisions uint64) { 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 replaces the state of the current view, so the - // view has to be stopped first, and no other view may start until the record - // is written. Otherwise a view could append a commit after that record and - // leave a sequence that recovery rejects. + // 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.viewTransitionLock.Lock() c.abortCurrentView() @@ -763,9 +823,11 @@ func (c *Controller) sync() (viewNum uint64, seq uint64, decisions uint64) { }, }, } - // A sync new view record is not a truncation point: the in-flight - // proposal, if there is one, must stay recoverable. - if err := c.State.Save(newViewToSave, false); err != nil { + // 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 + if err := c.State.Save(newViewToSave, truncateTo); err != nil { c.Logger.Panicf("Failed to save message to state, error: %v", err) } c.viewTransitionLock.Unlock() diff --git a/internal/bft/controller_sync_test.go b/internal/bft/controller_sync_test.go index f2637620..c64dd59e 100644 --- a/internal/bft/controller_sync_test.go +++ b/internal/bft/controller_sync_test.go @@ -6,13 +6,23 @@ 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" + "google.golang.org/protobuf/proto" ) // stubSynchronizer returns a fixed SyncResponse. @@ -290,7 +300,8 @@ func TestSyncAbortsViewBeforeSavingNewView(t *testing.T) { require.True(t, fixture.state.savedAfterAbort, "the new view was saved before the view stopped") require.Len(t, fixture.state.SavedMessages, 1) require.NotNil(t, fixture.state.SavedMessages[0].GetNewView()) - require.Equal(t, []bool{false}, fixture.state.TruncateTo) + require.Equal(t, []bool{true}, fixture.state.TruncateTo, + "a new view after sync truncates when no proposal is in flight") } // TestSyncHoldsViewTransitionLockUntilTheNewViewIsSaved covers the read side @@ -354,3 +365,397 @@ func TestSyncHoldsViewTransitionLockUntilTheNewViewIsSaved(t *testing.T) { t.Fatal("sync did not finish") } } + +// 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 +} + +// walSyncFixture 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 walSyncFixture struct { + controller *Controller + inFlight *InFlightData +} + +// newWALSyncFixture 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 newWALSyncFixture(t *testing.T, writeAheadLog api.WriteAheadLog, controllerSeq, controllerView, syncSeq, syncView uint64) *walSyncFixture { + 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 &walSyncFixture{controller: controller, inFlight: inFlight} +} + +// 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 *walSyncFixture) 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) AbortWithCancel(<-chan struct{}) { + p.Abort() +} + +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) +} + +// TestSyncAbortsViewBeforeWritingTheNewViewToWAL 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 TestSyncAbortsViewBeforeWritingTheNewViewToWAL(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 := newWALSyncFixture(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)) +} + +// 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. + controller.AbortView(1, nil) + 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 := newWALSyncFixture(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 := newWALSyncFixture(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 := newWALSyncFixture(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 86323e92..e700e3ff 100644 --- a/internal/bft/controller_test.go +++ b/internal/bft/controller_test.go @@ -76,6 +76,103 @@ 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) AbortWithCancel(<-chan struct{}) { + p.Abort() +} + +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, nil) + 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) @@ -1228,8 +1325,9 @@ func TestDeliverTwiceOnceFromSyncAndOnceFromViewData(t *testing.T) { return m.GetProposedRecord() != nil }), true).Return(nil) state.On("Save", mock.MatchedBy(func(m *protos.SavedMessage) bool { - return m.GetProposedRecord() == nil + return m.GetCommit() != nil }), false).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..90cdeff0 --- /dev/null +++ b/internal/bft/controller_view_guard_test.go @@ -0,0 +1,107 @@ +// Copyright IBM Corp. All Rights Reserved. +// +// SPDX-License-Identifier: Apache-2.0 +// + +package bft + +import ( + "sync" + "sync/atomic" + "testing" + + "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) AbortWithCancel(<-chan struct{}) { p.Abort() } +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") +} diff --git a/internal/bft/view.go b/internal/bft/view.go index 59e9a6a1..66ec470a 100644 --- a/internal/bft/view.go +++ b/internal/bft/view.go @@ -52,7 +52,7 @@ type State interface { // before the message can be dropped. The caller decides it; Save does not // infer the flag from the message type. A pre-prepare is normally a truncation // point, while a commit never is: its pre-prepare must survive next to it. - // View records are not truncation points in this change. + // A view record is a truncation point only when no proposal is in flight. Save(message *protos.SavedMessage, truncateTo bool) error // Restore restores the given view to its latest state diff --git a/internal/bft/view_test.go b/internal/bft/view_test.go index 82162b3e..23e9f797 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" @@ -1574,3 +1575,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 4137f5e7..48cb5b21 100644 --- a/internal/bft/viewchanger.go +++ b/internal/bft/viewchanger.go @@ -472,10 +472,10 @@ func (v *ViewChanger) processViewChangeMsg(restore bool) { }, }, } - // A view change record is not a truncation point: the proposal in - // flight, if there is one, travels in the view data message and must - // stay recoverable. - if err := v.State.Save(msgToSave, false); 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) } } @@ -498,6 +498,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() @@ -1235,16 +1242,24 @@ func (v *ViewChanger) processNewViewMsg(msg *protos.NewView) { return } - // The record of the new view replaces the state of the current view, so the view - // has to be stopped first. Otherwise it can still append a commit after that - // record and leave a sequence that recovery rejects. 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. + // 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.Controller.AbortViewAndHold(v.currView, v.stopChan) { 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{ @@ -1255,9 +1270,9 @@ func (v *ViewChanger) processNewViewMsg(msg *protos.NewView) { }, }, } - // A new view record is not a truncation point: the in-flight proposal, if - // there is one, must stay recoverable. - if err = v.State.Save(newViewToSave, false); 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.Controller.ReleaseViewWrite() diff --git a/internal/bft/viewchanger_test.go b/internal/bft/viewchanger_test.go index 201f07e7..981aacee 100644 --- a/internal/bft/viewchanger_test.go +++ b/internal/bft/viewchanger_test.go @@ -20,6 +20,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" @@ -173,7 +174,7 @@ func TestViewChangeProcess(t *testing.T) { controller := &mocks.ViewController{} expectAbortAndHold(controller) state := &mocks.State{} - state.On("Save", mock.Anything, false).Return(nil) + state.On("Save", mock.Anything, true).Return(nil) vc := &bft.ViewChanger{ SelfID: 0, @@ -279,7 +280,7 @@ func TestViewDataProcess(t *testing.T) { reqTimer := &mocks.RequestsTimer{} reqTimer.On("RestartTimers").Once() state := &mocks.State{} - state.On("Save", mock.Anything, false).Return(nil) + state.On("Save", mock.Anything, true).Return(nil) vc := &bft.ViewChanger{ SelfID: 1, @@ -326,7 +327,7 @@ func TestViewDataProcess(t *testing.T) { vc.Stop() reqTimer.AssertCalled(t, "RestartTimers") - state.AssertCalled(t, "Save", mock.Anything, false) + state.AssertCalled(t, "Save", mock.Anything, true) } func TestNewViewProcess(t *testing.T) { @@ -355,7 +356,7 @@ func TestNewViewProcess(t *testing.T) { reqTimer := &mocks.RequestsTimer{} reqTimer.On("RestartTimers").Once() state := &mocks.State{} - state.On("Save", mock.Anything, false).Return(nil) + state.On("Save", mock.Anything, true).Return(nil) vc := &bft.ViewChanger{ SelfID: 0, @@ -409,7 +410,7 @@ func TestNewViewProcess(t *testing.T) { vc.Stop() reqTimer.AssertCalled(t, "RestartTimers") - state.AssertCalled(t, "Save", mock.Anything, false) + state.AssertCalled(t, "Save", mock.Anything, true) } func TestNormalProcess(t *testing.T) { @@ -445,7 +446,7 @@ func TestNormalProcess(t *testing.T) { checkpoint := types.Checkpoint{} checkpoint.Set(lastDecision, lastDecisionSignatures) state := &mocks.State{} - state.On("Save", mock.Anything, false).Return(nil) + state.On("Save", mock.Anything, true).Return(nil) vc := &bft.ViewChanger{ SelfID: 1, @@ -1040,7 +1041,7 @@ func TestViewChangerTimeout(t *testing.T) { controllerWG := sync.WaitGroup{} controller.On("AbortView", mock.Anything, mock.Anything).Run(func(args mock.Arguments) { controllerWG.Done() - }) + }).Return(true) vc := &bft.ViewChanger{ N: 4, @@ -1098,7 +1099,7 @@ func TestBackOff(t *testing.T) { controllerWG := sync.WaitGroup{} controller.On("AbortView", mock.Anything, mock.Anything).Run(func(args mock.Arguments) { controllerWG.Done() - }) + }).Return(true) timeout := 10 * time.Second @@ -1183,7 +1184,7 @@ func TestCommitLastDecision(t *testing.T) { }).Return(types.Reconfig{InLatestDecision: false}) state := &mocks.State{} - state.On("Save", mock.Anything, false).Return(nil) + state.On("Save", mock.Anything, true).Return(nil) pruner := &mocks.Pruner{} pruner.On("MaybePruneRevokedRequests") @@ -1325,6 +1326,7 @@ func TestInFlightProposalInViewData(t *testing.T) { description string getInFlight func() *bft.InFlightData expectInflight bool + truncateTo bool }{ { description: "in flight is nil", @@ -1332,6 +1334,7 @@ func TestInFlightProposalInViewData(t *testing.T) { return &bft.InFlightData{} }, expectInflight: false, + truncateTo: true, }, { description: "in flight same as last decision", @@ -1341,6 +1344,7 @@ func TestInFlightProposalInViewData(t *testing.T) { return inFlight }, expectInflight: false, + truncateTo: false, }, { description: "in flight is after last decision", @@ -1355,6 +1359,7 @@ func TestInFlightProposalInViewData(t *testing.T) { return inFlight }, expectInflight: true, + truncateTo: false, }, } { t.Run(test.description, func(t *testing.T) { @@ -1381,7 +1386,7 @@ func TestInFlightProposalInViewData(t *testing.T) { checkpoint := types.Checkpoint{} checkpoint.Set(lastDecision, lastDecisionSignatures) state := &mocks.State{} - state.On("Save", mock.Anything, false).Return(nil) + state.On("Save", mock.Anything, test.truncateTo).Return(nil) vc := &bft.ViewChanger{ SelfID: 0, @@ -2108,7 +2113,6 @@ func TestCommitInFlight(t *testing.T) { histograms := &histogramsRecorder{} sched := make(chan time.Time) - state := &bft.StateRecorder{} vc := &bft.ViewChanger{ SelfID: 1, N: 4, @@ -2123,7 +2127,7 @@ func TestCommitInFlight(t *testing.T) { InFlight: &bft.InFlightData{}, Checkpoint: &checkpoint, ViewSequences: &atomic.Value{}, - State: state, + State: &bft.StateRecorder{}, InMsqQSize: int(types.DefaultConfig.IncomingMessageBufferSize), Application: app, Pruner: pruner, @@ -2219,9 +2223,6 @@ func TestCommitInFlight(t *testing.T) { } vc.Stop() - - require.Equal(t, []bool{false, false}, state.TruncateTo, - "a view change and a new view must not truncate the WAL in this change") } // histogramsRecorder is a metrics provider that records the values observed by its histograms. @@ -2539,7 +2540,7 @@ func TestDontCommitInFlight(t *testing.T) { app := &mocks.ApplicationMock{} app.On("Deliver", mock.Anything, mock.Anything) state := &mocks.State{} - state.On("Save", mock.Anything, false).Return(nil) + state.On("Save", mock.Anything, true).Return(nil) vc := &bft.ViewChanger{ SelfID: 3, @@ -2551,6 +2552,7 @@ func TestDontCommitInFlight(t *testing.T) { Ticker: make(chan time.Time), RequestsTimer: reqTimer, Application: app, + InFlight: &bft.InFlightData{}, State: state, } @@ -2632,7 +2634,7 @@ func newViewChangerWithInFlight(t *testing.T, verifier api.Verifier) (*bft.ViewC app := &mocks.ApplicationMock{} app.On("Deliver", mock.Anything, mock.Anything) state := &mocks.State{} - state.On("Save", mock.Anything, false).Return(nil) + state.On("Save", mock.Anything, true).Return(nil) vc := &bft.ViewChanger{ SelfID: 3, @@ -3052,7 +3054,7 @@ func TestNewViewWithoutLeaderViewData(t *testing.T) { pruner := &mocks.Pruner{} pruner.On("MaybePruneRevokedRequests") state := &mocks.State{} - state.On("Save", mock.Anything, false).Return(nil) + state.On("Save", mock.Anything, true).Return(nil) checkpoint := types.Checkpoint{} checkpoint.Set(lastDecision, lastDecisionSignatures) @@ -3314,3 +3316,419 @@ 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, mock.Anything) + controller.On("AbortViewAndHold", mock.Anything, mock.Anything).Return(true) + controller.On("ReleaseViewWrite") + 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, mock.Anything) + controller.On("AbortViewAndHold", mock.Anything, mock.Anything).Return(true) + controller.On("ReleaseViewWrite") + 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, mock.Anything) + controller.On("AbortViewAndHold", mock.Anything, mock.Anything).Return(true) + controller.On("ReleaseViewWrite") + 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, mock.Anything) + controller.On("AbortViewAndHold", mock.Anything, mock.Anything).Return(true) + controller.On("ReleaseViewWrite") + 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") + }) + } +} 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) + } + } +}