Restore the requests of a proposal recovered from the WAL

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

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)

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