feat: Consensus lag detection

Author: infrmtcs

consensus/driver/driver.go

@@ -176,11 +176,18 @@ func (d *Driver[V, H, A]) execute(
 			}
 
 			return true, nil
+
+		case *actions.TriggerSync:
+			d.triggerSync(*action)
 		}
 	}
 	return false, nil
 }
 
+func (d *Driver[V, H, A]) triggerSync(sync actions.TriggerSync) {
+	// TODO: Implement this
+}
+
 func (d *Driver[V, H, A]) setTimeout(ctx context.Context, timeout types.Timeout) {
 	d.scheduledTms[timeout] = time.AfterFunc(d.getTimeout(timeout.Step, timeout.Round), func() {
 		select {

consensus/tendermint/process.go

@@ -38,9 +38,36 @@ func (s *stateMachine[V, H, A]) ProcessPrecommit(p *types.Precommit[H, A]) []act
 	if !s.voteCounter.AddPrecommit(p) || !s.isHeightStarted {
 		return nil
 	}
+
+	if s.hasFuturePrecommitQuorum(p) {
+		return s.triggerSync(p)
+	}
+
 	return s.processMessage(p, (*wal.WALPrecommit[H, A])(p))
 }
 
+func (s *stateMachine[V, H, A]) hasFuturePrecommitQuorum(p *types.Precommit[H, A]) bool {
+	return p.ID != nil &&
+		p.Height > s.state.height &&
+		p.Height > s.lastTriggerSync &&
+		s.voteCounter.HasFuturePrecommitQuorum(p.Height, p.Round, p.ID)
+}
+
+func (s *stateMachine[V, H, A]) triggerSync(p *types.Precommit[H, A]) []actions.Action[V, H, A] {
+	syncStart := max(s.lastTriggerSync+1, s.state.height)
+	s.lastTriggerSync = p.Height
+
+	return []actions.Action[V, H, A]{
+		&actions.WriteWAL[V, H, A]{
+			Entry: (*wal.WALPrecommit[H, A])(p),
+		},
+		&actions.TriggerSync{
+			Start: syncStart,
+			End:   p.Height,
+		},
+	}
+}
+
 func (s *stateMachine[V, H, A]) processMessage(
 	msg types.Message[V, H, A],
 	walEntry wal.Entry[V, H, A],

consensus/tendermint/tendermint.go

@@ -40,6 +40,7 @@ type stateMachine[V types.Hashable[H], H types.Hash, A types.Addr] struct {
 	voteCounter     votecounter.VoteCounter[V, H, A]
 	application     Application[V, H]
 	isHeightStarted bool
+	lastTriggerSync types.Height
 }
 
 type state[V types.Hashable[H], H types.Hash] struct {

consensus/types/actions/actions.go

@@ -24,16 +24,22 @@ type ScheduleTimeout types.Timeout
 
 type Commit[V types.Hashable[H], H types.Hash, A types.Addr] types.Proposal[V, H, A]
 
+type TriggerSync struct {
+	Start, End types.Height
+}
+
 func (a *WriteWAL[V, H, A]) RequiresWALFlush() bool          { return false }
 func (a *BroadcastProposal[V, H, A]) RequiresWALFlush() bool { return true }
 func (a *BroadcastPrevote[H, A]) RequiresWALFlush() bool     { return true }
 func (a *BroadcastPrecommit[H, A]) RequiresWALFlush() bool   { return true }
 func (a *ScheduleTimeout) RequiresWALFlush() bool            { return false }
 func (a *Commit[V, H, A]) RequiresWALFlush() bool            { return true }
+func (a *TriggerSync) RequiresWALFlush() bool                { return false }
 
 func (a *WriteWAL[V, H, A]) isTendermintAction()          {}
 func (a *BroadcastProposal[V, H, A]) isTendermintAction() {}
 func (a *BroadcastPrevote[H, A]) isTendermintAction()     {}
 func (a *BroadcastPrecommit[H, A]) isTendermintAction()   {}
 func (a *ScheduleTimeout) isTendermintAction()            {}
 func (a *Commit[V, H, A]) isTendermintAction()            {}
+func (a *TriggerSync) isTendermintAction()                {}

consensus/votecounter/vote_counter.go

@@ -1,8 +1,6 @@
 package votecounter
 
 import (
-	"iter"
-
 	"github.com/NethermindEth/juno/consensus/types"
 	"github.com/NethermindEth/juno/utils"
 )
@@ -42,23 +40,6 @@ func New[V types.Hashable[H], H types.Hash, A types.Addr](validators Validators[
 	}
 }
 
-func (v *VoteCounter[V, H, A]) LoadFromMessages(messages iter.Seq2[types.Message[V, H, A], error]) error {
-	for msg, err := range messages {
-		if err != nil {
-			return err
-		}
-		switch msg := msg.(type) {
-		case *types.Proposal[V, H, A]:
-			v.AddProposal(msg)
-		case *types.Prevote[H, A]:
-			v.AddPrevote(msg)
-		case *types.Precommit[H, A]:
-			v.AddPrecommit(msg)
-		}
-	}
-	return nil
-}
-
 func (v *VoteCounter[V, H, A]) StartNewHeight() {
 	v.currentHeight++
 	v.totalVotingPower = v.validators.TotalVotingPower(v.currentHeight)
@@ -74,23 +55,26 @@ func (v *VoteCounter[V, H, A]) StartNewHeight() {
 	}
 }
 
-func (v *VoteCounter[V, H, A]) getRoundDataAndVotingPower(header *types.MessageHeader[A]) (*roundData[V, H, A], types.VotingPower, bool) {
-	if header.Height < v.currentHeight {
-		return nil, 0, false
+func (v *VoteCounter[V, H, A]) getRoundData(
+	height types.Height,
+	round types.Round,
+) (*roundData[V, H, A], bool) {
+	if height < v.currentHeight {
+		return nil, false
 	}
 
-	votingPower := v.validators.ValidatorVotingPower(header.Height, &header.Sender)
-
-	if header.Height == v.currentHeight {
-		return getOrCreateRoundData(v.roundData, header.Round), votingPower, true
+	var roundData roundMap[V, H, A]
+	if height == v.currentHeight {
+		roundData = v.roundData
+	} else {
+		var ok bool
+		if roundData, ok = v.futureMessages[height]; !ok {
+			roundData = make(roundMap[V, H, A])
+			v.futureMessages[height] = roundData
+		}
 	}
 
-	futureRoundMap, ok := v.futureMessages[header.Height]
-	if !ok {
-		futureRoundMap = make(roundMap[V, H, A])
-		v.futureMessages[header.Height] = futureRoundMap
-	}
-	return getOrCreateRoundData(futureRoundMap, header.Round), votingPower, true
+	return getOrCreateRoundData(roundData, round), true
 }
 
 func getOrCreateRoundData[V types.Hashable[H], H types.Hash, A types.Addr](
@@ -106,7 +90,7 @@ func getOrCreateRoundData[V types.Hashable[H], H types.Hash, A types.Addr](
 }
 
 func (v *VoteCounter[V, H, A]) AddProposal(proposal *types.Proposal[V, H, A]) bool {
-	roundData, votingPower, ok := v.getRoundDataAndVotingPower(&proposal.MessageHeader)
+	roundData, ok := v.getRoundData(proposal.Height, proposal.Round)
 	if !ok {
 		return false
 	}
@@ -115,24 +99,30 @@ func (v *VoteCounter[V, H, A]) AddProposal(proposal *types.Proposal[V, H, A]) bo
 		return false
 	}
 
+	votingPower := v.validators.ValidatorVotingPower(proposal.Height, &proposal.Sender)
+
 	return roundData.setProposal(proposal, votingPower)
 }
 
 func (v *VoteCounter[V, H, A]) AddPrevote(prevote *types.Prevote[H, A]) bool {
-	roundData, votingPower, ok := v.getRoundDataAndVotingPower(&prevote.MessageHeader)
+	roundData, ok := v.getRoundData(prevote.Height, prevote.Round)
 	if !ok {
 		return false
 	}
 
+	votingPower := v.validators.ValidatorVotingPower(prevote.Height, &prevote.Sender)
+
 	return roundData.addVote((*types.Vote[H, A])(prevote), votingPower, Prevote)
 }
 
 func (v *VoteCounter[V, H, A]) AddPrecommit(precommit *types.Precommit[H, A]) bool {
-	roundData, votingPower, ok := v.getRoundDataAndVotingPower(&precommit.MessageHeader)
+	roundData, ok := v.getRoundData(precommit.Height, precommit.Round)
 	if !ok {
 		return false
 	}
 
+	votingPower := v.validators.ValidatorVotingPower(precommit.Height, &precommit.Sender)
+
 	return roundData.addVote((*types.Vote[H, A])(precommit), votingPower, Precommit)
 }
 
@@ -163,6 +153,19 @@ func (v *VoteCounter[V, H, A]) HasQuorumForAny(round types.Round, voteType VoteT
 	return roundData.countAny(voteType) >= v.quorumVotingPower
 }
 
+func (v *VoteCounter[V, H, A]) HasFuturePrecommitQuorum(
+	height types.Height,
+	round types.Round,
+	id *H,
+) bool {
+	roundData, ok := v.getRoundData(height, round)
+	if !ok {
+		return false
+	}
+
+	return roundData.countVote(Precommit, id) >= v.quorumVotingPower
+}
+
 func (v *VoteCounter[V, H, A]) HasNonFaultyFutureMessage(round types.Round) bool {
 	roundData, ok := v.roundData[round]
 	if !ok {

consensus/votecounter/vote_counter_test.go

@@ -1,14 +1,12 @@
 package votecounter
 
 import (
-	"iter"
 	"testing"
 
 	"github.com/NethermindEth/juno/consensus/starknet"
 	"github.com/NethermindEth/juno/consensus/types"
 	"github.com/NethermindEth/juno/core/felt"
 	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
 )
 
 func assertVoteCounter(
@@ -23,12 +21,34 @@ func assertVoteCounter(
 	f := f(voteCounter.validators.TotalVotingPower(testHeight))
 	q := q(voteCounter.validators.TotalVotingPower(testHeight))
 
-	assert.Equal(t, countVotes(isVotingNil, isPrevote) >= q, voteCounter.HasQuorumForVote(testRound, Prevote, nil))
-	assert.Equal(t, countVotes(isVotingNil, isPrecommit) >= q, voteCounter.HasQuorumForVote(testRound, Precommit, nil))
+	assert.Equal(
+		t,
+		countVotes(isVotingNil, isPrevote) >= q,
+		voteCounter.HasQuorumForVote(testRound, Prevote, nil),
+	)
+	assert.Equal(
+		t,
+		countVotes(isVotingNil, isPrecommit) >= q,
+		voteCounter.HasQuorumForVote(testRound, Precommit, nil),
+	)
 
 	for _, id := range allIDs {
-		assert.Equal(t, countVotes(isVotingID(id), isPrevote) >= q, voteCounter.HasQuorumForVote(testRound, Prevote, &id))
-		assert.Equal(t, countVotes(isVotingID(id), isPrecommit) >= q, voteCounter.HasQuorumForVote(testRound, Precommit, &id))
+		assert.Equal(
+			t,
+			countVotes(isVotingID(id), isPrevote) >= q,
+			voteCounter.HasQuorumForVote(testRound, Prevote, &id),
+		)
+		assert.Equal(
+			t,
+			countVotes(isVotingID(id), isPrecommit) >= q,
+			voteCounter.HasQuorumForVote(testRound, Precommit, &id),
+		)
+
+		assert.Equal(
+			t,
+			countVotes(isVotingID(id), isPrecommit) >= q,
+			voteCounter.HasFuturePrecommitQuorum(testHeight+1, testRound, &id),
+		)
 	}
 
 	assert.Equal(t, countVotes(isPrevote) >= q, voteCounter.HasQuorumForAny(testRound, Prevote))
@@ -37,30 +57,6 @@ func assertVoteCounter(
 	assert.Equal(t, countVotes() > f, voteCounter.HasNonFaultyFutureMessage(testRound))
 }
 
-func convertToMessages(t *testing.T, testCases []testCase) iter.Seq2[starknet.Message, error] {
-	return func(yield func(starknet.Message, error) bool) {
-		t.Helper()
-		for _, testCase := range testCases {
-			var message starknet.Message
-			switch testCase := testCase.(type) {
-			case *proposalTestCase:
-				message = &testCase.proposal
-			case *voteTestCase:
-				switch testCase.voteType {
-				case Prevote:
-					message = (*starknet.Prevote)(&testCase.vote)
-				case Precommit:
-					message = (*starknet.Precommit)(&testCase.vote)
-				}
-			}
-			assert.NotNil(t, message)
-			if !yield(message, nil) {
-				return
-			}
-		}
-	}
-}
-
 func testMultipleHeights[T any](
 	t *testing.T,
 	message *T,
@@ -89,19 +85,6 @@ func testMultipleHeights[T any](
 func TestVoteCounter(t *testing.T) {
 	tests := simpleTestScenarios
 
-	t.Run("LoadFromMessages", func(t *testing.T) {
-		for i := range tests {
-			t.Run(tests[i].String(), func(t *testing.T) {
-				tests := tests[:i+1]
-				messages := convertToMessages(t, tests)
-				voteCounter := New[starknet.Value](simpleMockValidator, testHeight)
-
-				require.NoError(t, voteCounter.LoadFromMessages(messages))
-				assertVoteCounter(t, &voteCounter, tests, simpleVotingPower)
-			})
-		}
-	})
-
 	t.Run("Add", func(t *testing.T) {
 		var firstProposal *starknet.Proposal