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()