From 9ea3d07cb3641ddb0428c8072bbcbc30bcf4ec9f Mon Sep 17 00:00:00 2001 From: Hagar Meir Date: Thu, 15 Oct 2026 12:48:02 +0300 Subject: [PATCH] Remove the requests of a decided in-flight proposal from the pool When a view change commits an in-flight proposal, the view that decides it had no requests and no pre-prepare time. So its requests were never removed from the request pool (their timers kept running, leading to forwarding and complaints), CountTxsAll was not increased, and LatencyBatchProcessing recorded the time since the zero time. Set the in-flight view's requests from the verifier and its pre-prepare time right before it starts, and check both in TestCommitInFlight. Co-Authored-By: Claude Opus 5.5 (1M context) Signed-off-by: Hagar Meir --- internal/bft/viewchanger.go | 3 ++ internal/bft/viewchanger_test.go | 53 +++++++++++++++++++++++++++++++- 2 files changed, 55 insertions(+), 1 deletion(-) diff --git a/internal/bft/viewchanger.go b/internal/bft/viewchanger.go index 7e128dad..07951151 100644 --- a/internal/bft/viewchanger.go +++ b/internal/bft/viewchanger.go @@ -1292,6 +1292,8 @@ func (v *ViewChanger) commitInFlightProposal(proposal *protos.Proposal) (success Payload: proposal.Payload, Header: proposal.Header, } + // so that once the in flight proposal is decided its requests are removed from the pool + v.inFlightView.inFlightRequests = v.Verifier.RequestsFromProposal(*v.inFlightView.inFlightProposal) v.inFlightView.myProposalSig = v.Signer.SignProposal(*v.inFlightView.inFlightProposal, nil) v.inFlightView.lastBroadcastSent = &protos.Message{ Content: &protos.Message_Commit{ @@ -1312,6 +1314,7 @@ func (v *ViewChanger) commitInFlightProposal(proposal *protos.Proposal) (success <-v.Ticker <-v.Ticker + inFlightView.beginPrePrepare = time.Now() // the latency of processing the in flight proposal is measured from here inFlightView.Start() defer inFlightView.Abort() diff --git a/internal/bft/viewchanger_test.go b/internal/bft/viewchanger_test.go index eff0e8f3..c0410a3d 100644 --- a/internal/bft/viewchanger_test.go +++ b/internal/bft/viewchanger_test.go @@ -7,6 +7,7 @@ package bft_test import ( "errors" + "slices" "strings" "sync" "sync/atomic" @@ -16,6 +17,7 @@ import ( "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" "github.com/hyperledger/SmartBFT/pkg/metrics/disabled" "github.com/hyperledger/SmartBFT/pkg/types" protos "github.com/hyperledger/SmartBFT/smartbftprotos" @@ -1935,6 +1937,8 @@ func TestCommitInFlight(t *testing.T) { verifier := &mocks.VerifierMock{} verifier.On("VerifySignature", mock.Anything).Return(nil) verifier.On("VerifyConsenterSig", mock.Anything, mock.Anything).Return(nil, nil) + inFlightRequests := []types.RequestInfo{{ClientID: "1", ID: "1"}, {ClientID: "1", ID: "2"}} + verifier.On("RequestsFromProposal", mock.Anything).Return(inFlightRequests) controller := &mocks.ViewController{} viewNumChan := make(chan uint64) seqNumChan := make(chan uint64) @@ -1948,6 +1952,7 @@ func TestCommitInFlight(t *testing.T) { reqTimer := &mocks.RequestsTimer{} reqTimer.On("StopTimers") reqTimer.On("RestartTimers") + reqTimer.On("RemoveRequest", mock.Anything).Return(nil) checkpoint := types.Checkpoint{} checkpoint.Set(lastDecision, lastDecisionSignatures) app := &mocks.ApplicationMock{} @@ -1961,6 +1966,7 @@ func TestCommitInFlight(t *testing.T) { pruner := &mocks.Pruner{} pruner.On("MaybePruneRevokedRequests") + histograms := &histogramsRecorder{} sched := make(chan time.Time) vc := &bft.ViewChanger{ SelfID: 1, @@ -1981,7 +1987,7 @@ func TestCommitInFlight(t *testing.T) { Application: app, Pruner: pruner, MetricsBlacklist: api.NewMetricsBlacklist(&disabled.Provider{}), - MetricsView: api.NewMetricsView(&disabled.Provider{}), + MetricsView: api.NewMetricsView(histograms), } vc.Start(0) @@ -2060,10 +2066,55 @@ func TestCommitInFlight(t *testing.T) { controller.AssertNumberOfCalls(t, "AbortView", 2) 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 { + reqTimer.AssertCalled(t, "RemoveRequest", req) + } + // the time it took to process the in flight proposal is measured from when its view started + latencies := histograms.observed("view_latency_batch_processing") + assert.Len(t, latencies, 1) + for _, latency := range latencies { + assert.Less(t, latency, time.Minute.Seconds()) + } vc.Stop() } +// histogramsRecorder is a metrics provider that records the values observed by its histograms. +type histogramsRecorder struct { + disabled.Provider + lock sync.Mutex + values map[string][]float64 +} + +func (p *histogramsRecorder) NewHistogram(opts metrics.HistogramOpts) metrics.Histogram { + return &recordedHistogram{recorder: p, name: opts.Name} +} + +func (p *histogramsRecorder) observed(name string) []float64 { + p.lock.Lock() + defer p.lock.Unlock() + return slices.Clone(p.values[name]) +} + +type recordedHistogram struct { + recorder *histogramsRecorder + name string +} + +func (h *recordedHistogram) With(...string) metrics.Histogram { + return h +} + +func (h *recordedHistogram) Observe(value float64) { + h.recorder.lock.Lock() + defer h.recorder.lock.Unlock() + if h.recorder.values == nil { + h.recorder.values = make(map[string][]float64) + } + h.recorder.values[h.name] = append(h.recorder.values[h.name], value) +} + func TestCommitWhileHavingInFlight(t *testing.T) { comm := &mocks.CommMock{} msgChan := make(chan *protos.Message, 1)