Updating proposer calculator on restart (#1915)

* Updating proposer calculator on restart

* test fix

consensus/polybft/consensus_runtime_test.go

@@ -465,9 +465,10 @@ func Test_NewConsensusRuntime(t *testing.T) {
 	blockchainMock.On("GetStateProviderForBlock", mock.Anything).Return(new(stateProviderMock)).Once()
 	blockchainMock.On("GetSystemState", mock.Anything, mock.Anything).Return(systemStateMock).Once()
 	blockchainMock.On("GetHeaderByNumber", uint64(0)).Return(&types.Header{Number: 0, ExtraData: createTestExtraForAccounts(t, 0, validators, nil)})
+	blockchainMock.On("GetHeaderByNumber", uint64(1)).Return(&types.Header{Number: 1, ExtraData: createTestExtraForAccounts(t, 1, validators, nil)})
 
 	polybftBackendMock := new(polybftBackendMock)
-	polybftBackendMock.On("GetValidators", mock.Anything, mock.Anything).Return(validators).Twice()
+	polybftBackendMock.On("GetValidators", mock.Anything, mock.Anything).Return(validators).Times(3)
 
 	tmpDir := t.TempDir()
 	config := &runtimeConfig{

consensus/polybft/proposer_calculator.go

@@ -167,12 +167,23 @@ func NewProposerCalculator(config *runtimeConfig, logger hclog.Logger) (*Propose
 		return nil, err
 	}
 
-	return &ProposerCalculator{
+	pc := &ProposerCalculator{
 		snapshot: snap,
 		config:   config,
 		state:    config.State,
 		logger:   logger,
-	}, nil
+	}
+
+	// If the node was previously stopped, leaving the proposer calculator in an inconsistent state,
+	// proposer calculator needs to be updated.
+	blockNumber := config.blockchain.CurrentHeader().Number
+	if pc.snapshot.Height <= blockNumber {
+		if err = pc.update(blockNumber); err != nil {
+			return nil, err
+		}
+	}
+
+	return pc, nil
 }
 
 // NewProposerCalculator creates a new proposer calculator object
@@ -199,6 +210,11 @@ func (pc *ProposerCalculator) GetSnapshot() (*ProposerSnapshot, bool) {
 // It will update priorities and save the updated snapshot to db
 func (pc *ProposerCalculator) PostBlock(req *PostBlockRequest) error {
 	blockNumber := req.FullBlock.Block.Number()
+
+	return pc.update(blockNumber)
+}
+
+func (pc *ProposerCalculator) update(blockNumber uint64) error {
 	pc.logger.Debug("Update proposers snapshot started", "target block", blockNumber)
 
 	from := pc.snapshot.Height

e2e-polybft/e2e/consensus_test.go

@@ -5,6 +5,7 @@ import (
 	"math/big"
 	"path"
 	"strconv"
+	"sync"
 	"testing"
 	"time"
 
@@ -93,23 +94,33 @@ func TestE2E_Consensus_Basic_WithNonValidators(t *testing.T) {
 func TestE2E_Consensus_BulkDrop(t *testing.T) {
 	const (
 		clusterSize = 5
-		bulkToDrop  = 3
+		bulkToDrop  = 4
 		epochSize   = 5
 	)
 
 	cluster := framework.NewTestCluster(t, clusterSize,
-		framework.WithEpochSize(epochSize))
+		framework.WithEpochSize(epochSize),
+		framework.WithBlockTime(time.Second))
 	defer cluster.Stop()
 
 	// wait for cluster to start
 	cluster.WaitForReady(t)
 
+	var wg sync.WaitGroup
 	// drop bulk of nodes from cluster
 	for i := 0; i < bulkToDrop; i++ {
 		node := cluster.Servers[i]
-		node.Stop()
+
+		wg.Add(1)
+
+		go func(node *framework.TestServer) {
+			defer wg.Done()
+			node.Stop()
+		}(node)
 	}
 
+	wg.Wait()
+
 	// start dropped nodes again
 	for i := 0; i < bulkToDrop; i++ {
 		node := cluster.Servers[i]

e2e-polybft/framework/test-cluster.go

@@ -84,6 +84,7 @@ type TestClusterConfig struct {
 	LogsDir              string
 	TmpDir               string
 	BlockGasLimit        uint64
+	BlockTime            time.Duration
 	BurnContract         *polybft.BurnContractInfo
 	ValidatorPrefix      string
 	Binary               string
@@ -251,6 +252,12 @@ func WithEpochReward(epochReward int) ClusterOption {
 	}
 }
 
+func WithBlockTime(blockTime time.Duration) ClusterOption {
+	return func(h *TestClusterConfig) {
+		h.BlockTime = blockTime
+	}
+}
+
 func WithBlockGasLimit(blockGasLimit uint64) ClusterOption {
 	return func(h *TestClusterConfig) {
 		h.BlockGasLimit = blockGasLimit
@@ -467,6 +474,11 @@ func NewTestCluster(t *testing.T, validatorsCount int, opts ...ClusterOption) *T
 			"--trieroot", cluster.Config.InitialStateRoot.String(),
 		}
 
+		if cluster.Config.BlockTime != 0 {
+			args = append(args, "--block-time",
+				cluster.Config.BlockTime.String())
+		}
+
 		if cluster.Config.RelayerTrackerPollInterval != 0 {
 			args = append(args, "--block-tracker-poll-interval",
 				cluster.Config.RelayerTrackerPollInterval.String())