Prune the pool after a reconfig when the config is installed late

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)

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{})

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

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

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}
 		}