fix(state): proposer selected twice when prev block round > 0 (#904)

* chore: improve logging

* fix(state): invalid proposer selected if prev round > 0

* fix: correct validator increment

* test(consensus): TestStateProposerSelectionBetweenRoundsAndHeights

* build(deps): update go-deadlock to 0.3.5

* chore: builds

* chore: wip

* chore: wip

* WIP: removed score from validator - tests still red

* fix: quorum rotation

* chore: round score refactor

* chore: fix consensus

* chore: first proposer does not work

* fix: height score not resetting rounds

* fix typo

* test: remove old test

* chore: comments

* chore: moving some files around

* chore: move some files around

* chore: moving around a bit more

* chore: moving around

* chore: some renames

* chore: add logger as arg

* chore: comments

* chore: fix some lint issues

* chore: code cleanup

* chore: rename UpdateScores to UpdateHeightRound

* chore: remove unused SetLogger

* chore: MustGetProposer should only be used in tests

* test: fix some tests

* test: fix tests

* test: fix tests

* chore: some linter issues fixed

* chore: bump abci version

* chore: lint

* doc: add decision to ADR D002.

* doc: remove adr-d002

abci/example/kvstore/verify_test.go

@@ -85,7 +85,7 @@ func (e *blockExecutor) createBlock(txs types.Txs, commit *types.Commit) *types.
 	if commit == nil {
 		commit = &types.Commit{}
 	}
-	proposer := e.state.Validators.GetProposer()
+	proposer := e.state.GetProposerFromState(e.state.LastBlockHeight+1, 0)
 	block := e.state.MakeBlock(
 		e.state.LastBlockHeight+1,
 		txs,

dash/quorum/validator_conn_executor_test.go

@@ -621,7 +621,7 @@ func makeState(nVals int, height int64) (sm.State, dbm.DB) {
 }
 
 func makeBlock(ctx context.Context, t *testing.T, blockExec *sm.BlockExecutor, state sm.State, _height int64, commit *types.Commit) *types.Block {
-	block, crs, err := blockExec.CreateProposalBlock(ctx, 1, 0, state, commit, state.Validators.Proposer.ProTxHash, 1)
+	block, crs, err := blockExec.CreateProposalBlock(ctx, 1, 0, state, commit, state.Validators.Proposer().ProTxHash, 1)
 	require.NoError(t, err)
 
 	err = crs.UpdateBlock(block)

dash/quorum/validator_map_test.go

@@ -3,9 +3,10 @@ package quorum
 import (
 	"testing"
 
+	"github.com/stretchr/testify/assert"
+
 	"github.com/dashpay/tenderdash/dash/quorum/mock"
 	"github.com/dashpay/tenderdash/types"
-	"github.com/stretchr/testify/assert"
 )
 
 func Test_validatorMap_String(t *testing.T) {
@@ -30,11 +31,11 @@ func Test_validatorMap_String(t *testing.T) {
 		{
 			vm: newValidatorMap(vals),
 			contains: []string{
-				"<nil> VP:0 A:0 N:tcp://[email protected]:1}",
-				"<nil> VP:0 A:0 N:tcp://[email protected]:2}",
-				"<nil> VP:0 A:0 N:tcp://[email protected]:3}",
-				"<nil> VP:0 A:0 N:tcp://[email protected]:4}",
-				"<nil> VP:0 A:0 N:tcp://[email protected]:5}",
+				"<nil> VP:0 N:tcp://[email protected]:1}",
+				"<nil> VP:0 N:tcp://[email protected]:2}",
+				"<nil> VP:0 N:tcp://[email protected]:3}",
+				"<nil> VP:0 N:tcp://[email protected]:4}",
+				"<nil> VP:0 N:tcp://[email protected]:5}",
 			},
 		},
 	}

internal/consensus/gossiper.go

@@ -203,10 +203,10 @@ func (g *msgGossiper) GossipCommit(ctx context.Context, rs cstypes.RoundState, p
 	if prs.HasCommit {
 		return
 	}
-	logger := g.logger.With([]any{
+	logger := g.logger.With(
 		"height", rs.Height,
 		"peer_height", prs.Height,
-	})
+	)
 	var commit *types.Commit
 	blockStoreBase := g.blockStore.Base()
 	if rs.Height == prs.Height+1 {

internal/consensus/msgs_test.go

@@ -78,7 +78,7 @@ func TestMsgToProto(t *testing.T) {
 	vote, err := factory.MakeVote(
 		ctx,
 		pv,
-		&types.ValidatorSet{Proposer: val, Validators: []*types.Validator{val}, QuorumHash: quorumHash, ThresholdPublicKey: pk},
+		&types.ValidatorSet{Validators: []*types.Validator{val}, QuorumHash: quorumHash, ThresholdPublicKey: pk},
 		"chainID",
 		0,
 		1,

internal/consensus/pbts_test.go

@@ -293,7 +293,7 @@ type timestampedEvent struct {
 
 func (p *pbtsTestHarness) pickProposer() types.PrivValidator {
 	stateData := p.observedState.GetStateData()
-	proposer := stateData.Validators.GetProposer()
+	proposer := stateData.ProposerSelector.MustGetProposer(p.currentHeight, p.currentRound)
 	p.observedState.logger.Debug("picking proposer", "protxhash", proposer.ProTxHash)
 
 	allVals := append(p.otherValidators, p.observedValidator)

internal/consensus/peer_state.go

@@ -14,6 +14,7 @@ import (
 	cstypes "github.com/dashpay/tenderdash/internal/consensus/types"
 	"github.com/dashpay/tenderdash/libs/bits"
 	"github.com/dashpay/tenderdash/libs/log"
+	"github.com/dashpay/tenderdash/libs/math"
 	tmproto "github.com/dashpay/tenderdash/proto/tendermint/types"
 	"github.com/dashpay/tenderdash/types"
 )
@@ -214,7 +215,11 @@ func (ps *PeerState) PickVoteToSend(votes types.VoteSetReader) (*types.Vote, boo
 	}
 
 	if index, ok := votes.BitArray().Sub(psVotes).PickRandom(); ok {
-		vote := votes.GetByIndex(int32(index))
+		idx, err := math.SafeConvertInt32(int64(index))
+		if err != nil {
+			panic(fmt.Errorf("failed to convert index to int32: %w", err))
+		}
+		vote := votes.GetByIndex(idx)
 		if vote != nil {
 			return vote, true
 		}

internal/consensus/reactor.go

@@ -223,7 +223,7 @@ func (r *Reactor) SwitchToConsensus(ctx context.Context, state sm.State, skipWAL
 
 	// NOTE: The line below causes broadcastNewRoundStepRoutine() to broadcast a
 	// NewRoundStepMessage.
-	stateData.updateToState(state, nil)
+	stateData.updateToState(state, nil, r.state.blockStore)
 	err := r.state.stateDataStore.Update(stateData)
 	if err != nil {
 		panic(err)

internal/consensus/replay.go

@@ -74,7 +74,6 @@ func (cs *State) readReplayMessage(ctx context.Context, msg *TimedWALMessage, ne
 		case *ProposalMessage:
 			p := msg.Proposal
 			if cs.config.WalSkipRoundsToLast && p.Round > stateData.Round {
-				stateData.Validators.IncrementProposerPriority(p.Round - stateData.Round)
 				stateData.Votes.SetRound(p.Round)
 				stateData.Round = p.Round
 			}

internal/consensus/replay_test.go

@@ -495,7 +495,7 @@ func createSignSendProposal(ctx context.Context,
 	height := stateData.RoundState.Height
 	round := stateData.RoundState.Round
 
-	proposer := stateData.Validators.GetProposer()
+	proposer := stateData.ProposerSelector.MustGetProposer(height, round)
 	proposerVs := findValByProTxHash(ctx, t, vss, proposer.ProTxHash)
 	proposerCs := findStateByProTxHash(t, css, proposer.ProTxHash)
 

internal/consensus/replayer_test.go

@@ -188,7 +188,6 @@ func TestInitChainGenesisTime(t *testing.T) {
 	)
 	require.NoError(t, err)
 	vset.ThresholdPublicKey = recoveredThresholdPublicKey
-	proposerProTxHash := vset.GetProposer().ProTxHash
 
 	genDoc := tmtypes.GenesisDoc{
 		ChainID:            "test-chain",
@@ -208,6 +207,8 @@ func TestInitChainGenesisTime(t *testing.T) {
 	require.NoError(t, err)
 	stateStore := sm.NewStore(dbm.NewMemDB())
 	blockStore := store.NewBlockStore(dbm.NewMemDB())
+	proposer := smState.GetProposerFromState(1, 0)
+	proposerProTxHash := proposer.ProTxHash
 	replayer := newBlockReplayer(stateStore, blockStore, &genDoc, eventBus, proxyApp, proposerProTxHash)
 
 	// use replayer to call initChain

internal/consensus/state.go

@@ -334,7 +334,7 @@ func (cs *State) updateStateFromStore() error {
 		}
 	}
 
-	stateData.updateToState(state, nil)
+	stateData.updateToState(state, nil, cs.blockStore)
 	err = cs.stateDataStore.Update(stateData)
 	if err != nil {
 		return err

internal/consensus/state_apply_commit.go

@@ -34,9 +34,6 @@ func (c *ApplyCommitAction) Execute(ctx context.Context, stateEvent StateEvent)
 	event := stateEvent.Data.(*ApplyCommitEvent)
 	stateData := stateEvent.StateData
 	commit := event.Commit
-	c.logger.Info("applying commit", "commit", commit)
-
-	block, blockParts := stateData.ProposalBlock, stateData.ProposalBlockParts
 
 	height := stateData.Height
 	round := stateData.Round
@@ -45,6 +42,9 @@ func (c *ApplyCommitAction) Execute(ctx context.Context, stateEvent StateEvent)
 		height = commit.Height
 		round = commit.Round
 	}
+	c.logger.Info("applying commit", "commit", commit, "height", height, "round", round)
+
+	block, blockParts := stateData.ProposalBlock, stateData.ProposalBlockParts
 
 	c.blockExec.mustEnsureProcess(ctx, &stateData.RoundState, round)
 	c.blockExec.mustValidate(ctx, stateData)
@@ -88,7 +88,7 @@ func (c *ApplyCommitAction) Execute(ctx context.Context, stateEvent StateEvent)
 	c.RecordMetrics(stateData, height, block, lastBlockMeta)
 
 	// NewHeightStep!
-	stateData.updateToState(stateCopy, commit)
+	stateData.updateToState(stateCopy, commit, c.blockStore)
 	err = stateData.Save()
 	if err != nil {
 		return err

internal/consensus/state_data.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/dashpay/tenderdash/config"
 	cstypes "github.com/dashpay/tenderdash/internal/consensus/types"
+	selectproposer "github.com/dashpay/tenderdash/internal/consensus/versioned/selectproposer"
 	sm "github.com/dashpay/tenderdash/internal/state"
 	"github.com/dashpay/tenderdash/libs/eventemitter"
 	"github.com/dashpay/tenderdash/libs/log"
@@ -144,7 +145,12 @@ func (s *StateData) Save() error {
 }
 
 func (s *StateData) isProposer(proTxHash types.ProTxHash) bool {
-	return proTxHash != nil && bytes.Equal(s.Validators.GetProposer().ProTxHash.Bytes(), proTxHash.Bytes())
+	prop, err := s.ProposerSelector.GetProposer(s.Height, s.Round)
+	if err != nil {
+		s.logger.Error("error getting proposer", "err", err, "height", s.Height, "round", s.Round)
+		return false
+	}
+	return proTxHash != nil && bytes.Equal(prop.ProTxHash.Bytes(), proTxHash.Bytes())
 }
 
 func (s *StateData) isValidator(proTxHash types.ProTxHash) bool {
@@ -177,11 +183,17 @@ func (s *StateData) updateRoundStep(round int32, step cstypes.RoundStepType) {
 	}
 	s.Round = round
 	s.Step = step
+
+	if err := s.ProposerSelector.UpdateHeightRound(s.Height, round); err != nil {
+		s.logger.Error("error updating proposer scores",
+			"height", s.Height, "round", round,
+			"err", err)
+	}
 }
 
 // Updates State and increments height to match that of state.
 // The round becomes 0 and cs.Step becomes cstypes.RoundStepNewHeight.
-func (s *StateData) updateToState(state sm.State, commit *types.Commit) {
+func (s *StateData) updateToState(state sm.State, commit *types.Commit, blockStore selectproposer.BlockStore) {
 	if s.CommitRound > -1 && 0 < s.Height && s.Height != state.LastBlockHeight {
 		panic(fmt.Sprintf(
 			"updateToState() expected state height of %v but found %v",
@@ -221,7 +233,6 @@ func (s *StateData) updateToState(state sm.State, commit *types.Commit) {
 	}
 
 	// Reset fields based on state.
-	validators := state.Validators
 
 	switch {
 	case state.LastBlockHeight == 0: // Very first commit should be empty.
@@ -252,11 +263,31 @@ func (s *StateData) updateToState(state sm.State, commit *types.Commit) {
 		height = state.InitialHeight
 	}
 
-	s.logger.Trace("updating state height", "newHeight", height)
+	// state.Validators contain validator set at (state.LastBlockHeight+1, 0)
+	validators := state.Validators
 
-	// RoundState fields
-	s.updateHeight(height)
-	s.updateRoundStep(0, cstypes.RoundStepNewHeight)
+	if s.Validators == nil || !bytes.Equal(s.Validators.QuorumHash, validators.QuorumHash) {
+		s.logger.Info("Updating validators", "from", s.Validators.BasicInfoString(),
+			"to", validators.BasicInfoString())
+	}
+
+	s.Validators = validators
+	var err error
+
+	s.ProposerSelector, err = selectproposer.NewProposerSelector(
+		state.ConsensusParams,
+		s.Validators,
+		height,
+		0,
+		blockStore,
+		s.logger,
+	)
+	if err != nil {
+		s.logger.Error("error creating proposer selector", "height", height, "round", 0, "validators", s.Validators, "err", err)
+		panic(fmt.Sprintf("error creating proposer selector: %v", err))
+	}
+
+	s.logger.Trace("updating state height", "newHeight", height)
 
 	if s.CommitTime.IsZero() {
 		// "Now" makes it easier to sync up dev nodes.
@@ -265,12 +296,6 @@ func (s *StateData) updateToState(state sm.State, commit *types.Commit) {
 		s.StartTime = s.CommitTime
 	}
 
-	if s.Validators == nil || !bytes.Equal(s.Validators.QuorumHash, validators.QuorumHash) {
-		s.logger.Info("Updating validators", "from", s.Validators.BasicInfoString(),
-			"to", validators.BasicInfoString())
-	}
-
-	s.Validators = validators
 	s.Proposal = nil
 	s.ProposalReceiveTime = time.Time{}
 	s.ProposalBlock = nil
@@ -288,6 +313,10 @@ func (s *StateData) updateToState(state sm.State, commit *types.Commit) {
 	s.TriggeredTimeoutPrecommit = false
 
 	s.state = state
+
+	// RoundState fields
+	s.updateHeight(height)
+	s.updateRoundStep(0, cstypes.RoundStepNewHeight)
 }
 
 func (s *StateData) updateHeight(height int64) {

internal/consensus/state_enter_new_round.go

@@ -66,18 +66,10 @@ func (c *EnterNewRoundAction) Execute(ctx context.Context, stateEvent StateEvent
 		"round", stateData.Round,
 		"step", stateData.Step)
 
-	// increment validators if necessary
-	validators := stateData.Validators
-	if stateData.Round < round {
-		validators = validators.Copy()
-		validators.IncrementProposerPriority(round - stateData.Round)
-	}
-
 	// Setup new round
 	// we don't fire newStep for this step,
 	// but we fire an event, so update the round step first
 	stateData.updateRoundStep(round, cstypes.RoundStepNewRound)
-	stateData.Validators = validators
 	if round == 0 {
 		// We've already reset these upon new height,
 		// and meanwhile we might have received a proposal

internal/consensus/state_enter_propose.go

@@ -84,11 +84,14 @@ func (c *EnterProposeAction) Execute(ctx context.Context, stateEvent StateEvent)
 	c.scheduler.ScheduleTimeout(stateData.proposeTimeout(round), height, round, cstypes.RoundStepPropose)
 
 	if !isProposer {
+		prop, err := stateData.ProposerSelector.GetProposer(stateData.Height, stateData.Round)
+		if err != nil {
+			logger.Error("failed to get proposer", "err", err)
+			return nil // not a critical error, as we don't propose anyway
+		}
 		logger.Info("propose step; not our turn to propose",
-			"proposer_proTxHash", stateData.Validators.GetProposer().ProTxHash,
-			"node_proTxHash", proTxHash.String(),
-			"height", stateData.Height,
-			"round", stateData.Round,
+			"proposer_proTxHash", prop.ProTxHash.ShortString(),
+			"node_proTxHash", proTxHash.ShortString(),
 			"step", stateData.Step)
 		return nil
 	}
@@ -99,9 +102,8 @@ func (c *EnterProposeAction) Execute(ctx context.Context, stateEvent StateEvent)
 	}
 
 	logger.Info("propose step; our turn to propose",
+		"node_proTxHash", proTxHash.ShortString(),
 		"proposer_proTxHash", proTxHash.ShortString(),
-		"height", stateData.Height,
-		"round", stateData.Round,
 		"step", stateData.Step,
 	)
 	// Flush the WAL. Otherwise, we may not recompute the same proposal to sign,