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