From a87102a9da26a4eb9feeee21012c91290c81231c Mon Sep 17 00:00:00 2001 From: Hagar Meir Date: Thu, 22 Oct 2026 14:53:21 +0300 Subject: [PATCH] Prune the pool after a reconfig when the config is installed late Consensus.reconfig prunes the request pool and then starts the new controller, and Controller.Start stored the verification sequence the Verifier reported at that moment. If the application installs the new config asynchronously and the switch lands during the prune or before Controller.Start, the prune used the old config while the controller stored the new sequence. MaybePruneRevokedRequests then never saw a change, and requests that are invalid under the new config stayed in the pool. Read the verification sequence before the prune in reconfig and start the controller with it, so a switch that lands later is detected and pruned on the next decision, sync or view change. The initial Start passes the Verifier's current sequence, as before. The test app can now delay revoking clients until the prune during reconfig, and TestPoolPrunedAfterAsyncReconfig checks that the next decision removes the revoked request. Fixes #728 Written by Claude (AI assistant) on behalf of Hagar Meir. Co-Authored-By: Claude Opus 5.5 (1M context) Signed-off-by: Hagar Meir --- internal/bft/controller.go | 8 ++-- internal/bft/controller_test.go | 22 +++++----- pkg/consensus/consensus.go | 16 ++++--- test/reconfig_test.go | 76 +++++++++++++++++++++++++++++++++ test/test_app.go | 17 ++++++-- 5 files changed, 116 insertions(+), 23 deletions(-) diff --git a/internal/bft/controller.go b/internal/bft/controller.go index 2e53fbd9..e04c88e1 100644 --- a/internal/bft/controller.go +++ b/internal/bft/controller.go @@ -802,8 +802,10 @@ func (c *Controller) syncOnStart(startViewNumber uint64, startProposalSequence u return viewNum, seq, decisions } -// Start the controller -func (c *Controller) Start(startViewNumber uint64, startProposalSequence uint64, startDecisionsInView uint64, syncOnStart bool) { +// Start the controller. +// The verificationSequence is the one the requests in the pool were last pruned against, +// so that a later change of the Verifier's sequence prunes the pool again. +func (c *Controller) Start(startViewNumber uint64, startProposalSequence uint64, startDecisionsInView uint64, verificationSequence uint64, syncOnStart bool) { c.Logger.Debugf("Starting controller with view %d, sequence %d, and decisions %d", startViewNumber, startProposalSequence, startDecisionsInView) c.controllerDone.Add(1) c.stopOnce = sync.Once{} @@ -818,7 +820,7 @@ func (c *Controller) Start(startViewNumber uint64, startProposalSequence uint64, c.Logger.Debugf("The number of nodes (N) is %d, F is %d, and the quorum size is %d", c.N, F, Q) c.quorum = Q - c.verificationSequence.Store(c.Verifier.VerificationSequence()) + c.verificationSequence.Store(verificationSequence) if syncOnStart { startViewNumber, startProposalSequence, startDecisionsInView = c.syncOnStart(startViewNumber, startProposalSequence, startDecisionsInView) diff --git a/internal/bft/controller_test.go b/internal/bft/controller_test.go index 692cd3fa..548099eb 100644 --- a/internal/bft/controller_test.go +++ b/internal/bft/controller_test.go @@ -64,7 +64,7 @@ func TestControllerBasic(t *testing.T) { configureProposerBuilder(controller) - controller.Start(1, 0, 0, false) + controller.Start(1, 0, 0, 1, false) leaderMon.On("ProcessMsg", uint64(2), heartbeat) controller.ProcessMessages(2, heartbeat) @@ -121,7 +121,7 @@ func TestControllerLeaderBasic(t *testing.T) { configureProposerBuilder(controller) - controller.Start(1, 0, 0, false) + controller.Start(1, 0, 0, 1, false) <-batcherChan controller.Stop() batcher.AssertCalled(t, "NextBatch") @@ -230,7 +230,7 @@ func TestLeaderPropose(t *testing.T) { configureProposerBuilder(controller) commWG.Add(9) - controller.Start(1, 0, 0, true) + controller.Start(1, 0, 0, 1, true) commWG.Wait() // propose + state request commWG.Add(3) @@ -364,7 +364,7 @@ func TestViewChanged(t *testing.T) { configureProposerBuilder(controller) commWG.Add(3) // state request - controller.Start(1, 0, 0, true) + controller.Start(1, 0, 0, 1, true) commWG.Wait() commWG.Add(6) // propose @@ -472,7 +472,7 @@ func TestSyncPrevView(t *testing.T) { controller.ViewSequences = vs leaderMonWG.Add(1) - controller.Start(1, 0, 0, false) + controller.Start(1, 0, 0, 1, false) leaderMonWG.Wait() appWG.Add(1) @@ -640,7 +640,7 @@ func TestControllerLeaderRequestHandling(t *testing.T) { controller.Deliver = &bft.MutuallyExclusiveDeliver{C: controller} configureProposerBuilder(controller) - controller.Start(testCase.startViewNum, 0, 0, true) + controller.Start(testCase.startViewNum, 0, 0, 0, true) controller.HandleRequest(3, []byte{1, 2, 3}) @@ -830,7 +830,7 @@ func TestSyncInform(t *testing.T) { synchronizerWG.Add(1) commWG.Add(9) - controller.Start(1, 0, 0, true) + controller.Start(1, 0, 0, 1, true) synchronizerWG.Wait() commWG.Wait() @@ -943,7 +943,7 @@ func TestRotateFromLeaderToFollower(t *testing.T) { controller.ViewSequences = vs commWG.Add(6) - controller.Start(1, 0, 0, false) + controller.Start(1, 0, 0, 1, false) commWG.Wait() // propose (pre-prepare + prepare) commWG.Add(3) @@ -1125,7 +1125,7 @@ func TestRotateFromFollowerToLeader(t *testing.T) { controller.ViewSequences = vs followerMonWG.Add(1) // change role - controller.Start(1, 0, 0, false) + controller.Start(1, 0, 0, 1, false) followerMonWG.Wait() commWG.Add(3) @@ -1358,7 +1358,7 @@ func TestDeliverTwiceOnceFromSyncAndOnceFromViewData(t *testing.T) { controller.ViewSequences = vs leaderMonWG.Add(1) - controller.Start(0, 1, 0, false) + controller.Start(0, 1, 0, 1, false) leaderMonWG.Wait() // Starting view change @@ -1610,7 +1610,7 @@ func TestControllerKeepsRunningAfterDecideWaiterLeft(t *testing.T) { StartedWG: &startedWG, } - controller.Start(0, 1, 0, false) + controller.Start(0, 1, 0, 1, false) view := <-views decided := make(chan struct{}) diff --git a/pkg/consensus/consensus.go b/pkg/consensus/consensus.go index 2076940d..fe01e589 100644 --- a/pkg/consensus/consensus.go +++ b/pkg/consensus/consensus.go @@ -157,7 +157,7 @@ func (c *Consensus) Start() error { go c.run() - c.startComponents(view, seq, dec, true) + c.startComponents(view, seq, dec, c.Verifier.VerificationSequence(), true) atomic.StoreUint64(&c.running, 1) @@ -230,7 +230,11 @@ func (c *Consensus) reconfig(reconfig types.Reconfig) { c.continueCreateComponents() // Prune requests that are invalid under the new configuration now, while all components are stopped. - c.Logger.Infof("Pruning the request pool after reconfig") + // The application may install the new configuration asynchronously, so read the verification sequence + // before pruning and start the controller with it; if the sequence changes during or after the prune, + // the controller then sees the change and prunes again. + verificationSeq := c.Verifier.VerificationSequence() + c.Logger.Infof("Pruning the request pool after reconfig with verification sequence %d", verificationSeq) c.Pool.Prune(func(req []byte) error { _, err := c.Verifier.VerifyRequest(req) return err @@ -247,7 +251,7 @@ func (c *Consensus) reconfig(reconfig types.Reconfig) { c.waitForEachOther() - c.startComponents(view, seq, dec, false) + c.startComponents(view, seq, dec, verificationSeq, false) c.Pool.RestartTimers() @@ -512,14 +516,14 @@ func (c *Consensus) waitForEachOther() { c.controller.StartedWG = &c.viewChanger.ControllerStartedWG } -func (c *Consensus) startComponents(view, seq, dec uint64, configSync bool) { +func (c *Consensus) startComponents(view, seq, dec, verificationSeq uint64, configSync bool) { // If we delivered to the application proposal with sequence i, // then we are expecting to be proposed a proposal with sequence i+1. c.collector.Start() c.viewChanger.Start(view) if configSync { - c.controller.Start(view, seq+1, dec, c.Config.SyncOnStart) + c.controller.Start(view, seq+1, dec, verificationSeq, c.Config.SyncOnStart) } else { - c.controller.Start(view, seq+1, dec, false) + c.controller.Start(view, seq+1, dec, verificationSeq, false) } } diff --git a/test/reconfig_test.go b/test/reconfig_test.go index a55fccdf..bbed929f 100644 --- a/test/reconfig_test.go +++ b/test/reconfig_test.go @@ -138,6 +138,82 @@ func TestPoolPrunedAfterReconfig(t *testing.T) { } } +func TestPoolPrunedAfterAsyncReconfig(t *testing.T) { + // The follower installs the new config while the request pool is pruned during reconfig, + // so that prune still sees the old config and keeps a request that becomes invalid. + // The request must then be pruned on the next decision. + + t.Parallel() + network := NewNetwork() + defer network.Shutdown() + + testDir := t.TempDir() + + // Long enough for the follower to never forward the request to the leader during the test + forwardTimeout := time.Minute + + numberOfNodes := 4 + nodes := make([]*App, 0) + for i := 1; i <= numberOfNodes; i++ { + n := newNode(uint64(i), network, t.Name(), testDir, false, 0) + n.Consensus.Config.RequestForwardTimeout = forwardTimeout + n.Consensus.Config.RequestComplainTimeout = 2 * forwardTimeout + n.clientsToRevoke = []string{"bob"} + nodes = append(nodes, n) + } + + follower := nodes[1] + follower.revokeDuringPrune = true + + startNodes(nodes, network) + + follower.Submit(Request{ID: "1", ClientID: "bob"}) + require.Equal(t, 1, follower.Consensus.Pool.Size()) + + newConfig := fastConfig + newConfig.RequestForwardTimeout = forwardTimeout + newConfig.RequestComplainTimeout = 2 * forwardTimeout + + nodes[0].Submit(Request{ + ClientID: "reconfig", + ID: "10", + Reconfig: Reconfig{ + InLatestDecision: true, + CurrentNodes: nodesToInt(nodes[0].Node.Nodes()), + CurrentConfig: recconfigToInt(types.Reconfig{CurrentConfig: newConfig}).CurrentConfig, + }, + }) + + data := make([]*AppRecord, 0) + for i := range numberOfNodes { + d := <-nodes[i].Delivered + data = append(data, d) + } + for i := range numberOfNodes - 1 { + require.Equal(t, data[i], data[i+1]) + } + + // The prune during reconfig installed the new config but kept the request + require.Eventually(t, func() bool { + return follower.VerificationSequence() == 1 + }, 10*time.Second, 100*time.Millisecond) + require.Equal(t, 1, follower.Consensus.Pool.Size()) + + nodes[0].Submit(Request{ID: "11", ClientID: "alice"}) + data = make([]*AppRecord, 0) + for i := range numberOfNodes { + d := <-nodes[i].Delivered + data = append(data, d) + } + for i := range numberOfNodes - 1 { + require.Equal(t, data[i], data[i+1]) + } + + require.Eventually(t, func() bool { + return follower.Consensus.Pool.Size() == 0 + }, 10*time.Second, 100*time.Millisecond) +} + func TestBasicAddNodes(t *testing.T) { t.Parallel() network := NewNetwork() diff --git a/test/test_app.go b/test/test_app.go index 3168519f..b495954b 100644 --- a/test/test_app.go +++ b/test/test_app.go @@ -65,8 +65,12 @@ type App struct { verificationSeq uint64 clientsToRevoke []string // clients whose requests become invalid once a reconfig is delivered revokedClients sync.Map - messageLost func(*smartbftprotos.Message) bool - lock sync.Mutex + // revokeDuringPrune delays revoking the clients after a reconfig is delivered until the first request + // is verified afterwards, which models an application that installs the new config asynchronously + revokeDuringPrune bool + revokePending atomic.Bool + messageLost func(*smartbftprotos.Message) bool + lock sync.Mutex } type lastRecord struct { @@ -235,6 +239,9 @@ func (a *App) VerifyRequest(val []byte) (types.RequestInfo, error) { if _, revoked := a.revokedClients.Load(req.ClientID); revoked { return types.RequestInfo{}, fmt.Errorf("client %s is revoked", req.ClientID) } + if a.revokePending.CompareAndSwap(true, false) { + a.revokeClients() + } return types.RequestInfo{ID: req.ID, ClientID: req.ClientID}, nil } @@ -321,7 +328,11 @@ func (a *App) Deliver(proposal types.Proposal, signatures []types.Signature) typ for _, req := range record.Batch.Requests { request := requestFromBytes(req) if request.Reconfig.InLatestDecision { - a.revokeClients() + if a.revokeDuringPrune { + a.revokePending.Store(true) + } else { + a.revokeClients() + } reconfig := request.Reconfig.recconfigToUint(a.ID) return types.Reconfig{InLatestDecision: true, CurrentNodes: reconfig.CurrentNodes, CurrentConfig: reconfig.CurrentConfig} }