Prune the request pool after a reconfig

pkg/consensus/consensus.go

@@ -229,6 +229,13 @@ func (c *Consensus) reconfig(reconfig types.Reconfig) {
 	c.Pool.ChangeOptions(c.controller, opts) // TODO handle reconfiguration of queue size in the pool
 	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")
+	c.Pool.Prune(func(req []byte) error {
+		_, err := c.Verifier.VerifyRequest(req)
+		return err
+	})
+
 	proposal, _ := c.checkpoint.Get()
 	md := &protos.ViewMetadata{}
 	if err := proto.Unmarshal(proposal.Metadata, md); err != nil {

test/reconfig_test.go

@@ -75,6 +75,74 @@ func TestBasicReconfig(t *testing.T) {
 	}
 }
 
+func TestPoolPrunedAfterReconfig(t *testing.T) {
+	// A follower holds a request in its pool that becomes invalid once a reconfig is delivered,
+	// so the request must be pruned from the pool after the reconfig.
+
+	t.Parallel()
+	network := NewNetwork()
+	defer network.Shutdown()
+
+	testDir, err := os.MkdirTemp("", t.Name())
+	assert.NoErrorf(t, err, "generate temporary test dir")
+	defer os.RemoveAll(testDir)
+
+	// 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)
+	}
+	startNodes(nodes, network)
+
+	follower := nodes[1]
+	follower.Submit(Request{ID: "1", ClientID: "bob"})
+	assert.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 := 0; i < numberOfNodes-1; i++ {
+		assert.Equal(t, data[i], data[i+1])
+	}
+
+	assert.Eventually(t, func() bool {
+		return follower.Consensus.Pool.Size() == 0
+	}, 10*time.Second, 100*time.Millisecond)
+
+	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 := 0; i < numberOfNodes-1; i++ {
+		assert.Equal(t, data[i], data[i+1])
+	}
+}
+
 func TestBasicAddNodes(t *testing.T) {
 	t.Parallel()
 	network := NewNetwork()

test/test_app.go

@@ -63,6 +63,8 @@ type App struct {
 	metricsProvider metrics.Provider
 	lastRecord      lastRecord
 	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
 }
@@ -230,6 +232,9 @@ func (a *App) RequestsFromProposal(proposal types.Proposal) []types.RequestInfo
 // VerifyRequest verifies the given request and returns its info
 func (a *App) VerifyRequest(val []byte) (types.RequestInfo, error) {
 	req := requestFromBytes(val)
+	if _, revoked := a.revokedClients.Load(req.ClientID); revoked {
+		return types.RequestInfo{}, fmt.Errorf("client %s is revoked", req.ClientID)
+	}
 	return types.RequestInfo{ID: req.ID, ClientID: req.ClientID}, nil
 }
 
@@ -316,6 +321,7 @@ 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()
 			reconfig := request.Reconfig.recconfigToUint(a.ID)
 			return types.Reconfig{InLatestDecision: true, CurrentNodes: reconfig.CurrentNodes, CurrentConfig: reconfig.CurrentConfig}
 		}
@@ -324,6 +330,17 @@ func (a *App) Deliver(proposal types.Proposal, signatures []types.Signature) typ
 	return types.Reconfig{InLatestDecision: false}
 }
 
+// revokeClients revokes the clients in clientsToRevoke and advances the verification sequence
+func (a *App) revokeClients() {
+	if len(a.clientsToRevoke) == 0 {
+		return
+	}
+	for _, client := range a.clientsToRevoke {
+		a.revokedClients.Store(client, struct{}{})
+	}
+	atomic.AddUint64(&a.verificationSeq, 1)
+}
+
 type committedBatches struct {
 	lock     sync.RWMutex
 	latestMD *smartbftprotos.ViewMetadata