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