From 93149c38954d4cd5eff77cec1bcac89f4e886e45 Mon Sep 17 00:00:00 2001 From: Hagar Meir Date: Sat, 10 Oct 2026 22:25:25 +0300 Subject: [PATCH] Prune the request pool after a reconfig After a reconfig the new controller initializes its verification sequence from the verifier, which already reflects the new config, so MaybePruneRevokedRequests never detects the change. The prune in the old controller's decide is usually skipped because the controller is closed by then. As a result, requests revoked by the new config remained in the pool. Prune the pool unconditionally in Consensus.reconfig, after the new components are created and before they are started. Fixes #698 Co-Authored-By: Claude Opus 5.5 (1M context) Signed-off-by: Hagar Meir --- pkg/consensus/consensus.go | 7 ++++ test/reconfig_test.go | 68 ++++++++++++++++++++++++++++++++++++++ test/test_app.go | 17 ++++++++++ 3 files changed, 92 insertions(+) diff --git a/pkg/consensus/consensus.go b/pkg/consensus/consensus.go index 7321378e..9e370e57 100644 --- a/pkg/consensus/consensus.go +++ b/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 { diff --git a/test/reconfig_test.go b/test/reconfig_test.go index 8dd61aba..843f217b 100644 --- a/test/reconfig_test.go +++ b/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() diff --git a/test/test_app.go b/test/test_app.go index e36df67c..9740d69f 100644 --- a/test/test_app.go +++ b/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