From 777f8ce126d6c7216f406041c4ece1d06ffa5fad Mon Sep 17 00:00:00 2001 From: Hagar Meir Date: Tue, 20 Oct 2026 16:09:41 +0300 Subject: [PATCH] Restore the requests of a proposal recovered from the WAL A view restored from the WAL after a restart had its proposal but not its requests, and no pre-prepare time. So when it decided the proposal, its requests were not removed from the pool, CountTxsAll was not increased, and LatencyBatchProcessing recorded the time since the zero time. Set the view's requests from the verifier and its pre-prepare time when restoring an in-flight proposal, as a view does when it accepts one, and check both in TestViewPersisted. Co-Authored-By: Claude Opus 5.5 (1M context) Signed-off-by: Hagar Meir --- internal/bft/state.go | 11 +++++++++++ internal/bft/state_test.go | 5 +++++ internal/bft/view_test.go | 21 ++++++++++++++++++--- 3 files changed, 34 insertions(+), 3 deletions(-) diff --git a/internal/bft/state.go b/internal/bft/state.go index ffe9ab21..6743b4d1 100644 --- a/internal/bft/state.go +++ b/internal/bft/state.go @@ -8,6 +8,7 @@ package bft import ( "errors" "fmt" + "time" "github.com/hyperledger/SmartBFT/pkg/api" "github.com/hyperledger/SmartBFT/pkg/types" @@ -160,6 +161,7 @@ func (ps *PersistedState) recoverProposed(lastPersistedMessage *protos.ProposedR Payload: prop.Payload, Header: prop.Header, } + restoreInFlightRequests(v) ps.storeProposal(lastPersistedMessage) // Reconstruct the prepare message we shall next broadcast // after the recovery. @@ -219,6 +221,7 @@ func (ps *PersistedState) recoverPrepared(lastPersistedMessage *protos.Message, Payload: prop.Payload, Header: prop.Header, } + restoreInFlightRequests(v) ps.storeProposal(prePrepareMsg.GetProposedRecord()) ps.storePrepared(lastPersistedMessage) @@ -245,3 +248,11 @@ func (ps *PersistedState) recoverPrepared(lastPersistedMessage *protos.Message, ps.Logger.Infof("Restored proposal with sequence %d", prePrepareFromWAL.Seq) return nil } + +// restoreInFlightRequests sets what a view sets when it accepts a proposal and that is not persisted in the WAL, +// so that once the restored proposal is decided its requests are removed from the pool, +// and the time it took to process it is measured from the restore. +func restoreInFlightRequests(v *View) { + v.inFlightRequests = v.Verifier.RequestsFromProposal(*v.inFlightProposal) + v.beginPrePrepare = time.Now() +} diff --git a/internal/bft/state_test.go b/internal/bft/state_test.go index 64a6f6f0..22a5cc5c 100644 --- a/internal/bft/state_test.go +++ b/internal/bft/state_test.go @@ -9,11 +9,13 @@ import ( "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" protos "github.com/hyperledger/SmartBFT/smartbftprotos" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" "go.uber.org/zap" ) @@ -179,10 +181,13 @@ func TestStateRestore(t *testing.T) { InFlightProposal: &bft.InFlightData{}, } + verifier := &mocks.VerifierMock{} + verifier.On("RequestsFromProposal", mock.Anything).Return(nil) view := &bft.View{ Number: 300, ProposalSequence: testCase.proposalSeqViewInitializedWith, MetricsView: api.NewMetricsView(&disabled.Provider{}), + Verifier: verifier, } err = state.Restore(view) diff --git a/internal/bft/view_test.go b/internal/bft/view_test.go index 1f0130f3..c6883d4d 100644 --- a/internal/bft/view_test.go +++ b/internal/bft/view_test.go @@ -13,6 +13,7 @@ import ( "sync" "sync/atomic" "testing" + "time" "github.com/hyperledger/SmartBFT/internal/bft" "github.com/hyperledger/SmartBFT/internal/bft/mocks" @@ -920,10 +921,12 @@ func TestViewPersisted(t *testing.T) { }, } { t.Run(testCase.description, func(t *testing.T) { + requests := []types.RequestInfo{{ClientID: "1", ID: "1"}, {ClientID: "1", ID: "2"}} verifier := &mocks.VerifierMock{} verifier.On("VerifySignature", mock.Anything).Return(nil) verifier.On("VerificationSequence").Return(uint64(1)) - verifier.On("VerifyProposal", mock.Anything).Return(nil, nil) + verifier.On("VerifyProposal", mock.Anything).Return(requests, nil) + verifier.On("RequestsFromProposal", mock.Anything).Return(requests).Maybe() verifier.On("VerifyConsenterSig", mock.Anything, mock.Anything).Return(nil, nil) var prepareSent sync.WaitGroup @@ -958,10 +961,13 @@ func TestViewPersisted(t *testing.T) { var deciderWG sync.WaitGroup deciderWG.Add(1) + var decidedRequests []types.RequestInfo decider := &mocks.Decider{} - decider.On("Decide", mock.Anything, mock.Anything, mock.Anything).Run(func(_ mock.Arguments) { + decider.On("Decide", mock.Anything, mock.Anything, mock.Anything).Run(func(args mock.Arguments) { + decidedRequests = args.Get(2).([]types.RequestInfo) deciderWG.Done() }) + histograms := &histogramsRecorder{} state := &mocks.State{} @@ -983,7 +989,7 @@ func TestViewPersisted(t *testing.T) { ProposalSequence: 0, ViewSequences: &atomic.Value{}, InMsgQSize: 40, - MetricsView: api.NewMetricsView(&disabled.Provider{}), + MetricsView: api.NewMetricsView(histograms), } } @@ -1095,6 +1101,15 @@ func TestViewPersisted(t *testing.T) { // Wait for the proposal to be committed. deciderWG.Wait() + // The requests of the proposal are decided with it, also when the view was restored from the WAL. + assert.Equal(t, requests, decidedRequests) + // The time it took to process the proposal is measured from when the view got it, or was restored with it. + latencies := histograms.observed("view_latency_batch_processing") + assert.Len(t, latencies, 1) + for _, latency := range latencies { + assert.Less(t, latency, time.Minute.Seconds()) + } + view.Abort() writeAheadLog.Close()