diff --git a/sei-tendermint/internal/autobahn/consensus/commit_votes.go b/sei-tendermint/internal/autobahn/consensus/commit_votes.go deleted file mode 100644 index 047add000c..0000000000 --- a/sei-tendermint/internal/autobahn/consensus/commit_votes.go +++ /dev/null @@ -1,68 +0,0 @@ -package consensus - -import ( - "github.com/sei-protocol/sei-chain/sei-tendermint/autobahn/types" - "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils" -) - -type scv = *types.Signed[*types.CommitVote] -type hcv = types.Hash[*types.CommitVote] - -type commitVotes struct { - byKey map[types.PublicKey]scv - byHash map[hcv]*voteSet[scv] - qc utils.AtomicSend[utils.Option[*types.CommitQC]] -} - -func newCommitVotes() *commitVotes { - return &commitVotes{ - byKey: map[types.PublicKey]scv{}, - byHash: map[hcv]*voteSet[scv]{}, - qc: utils.NewAtomicSend(utils.None[*types.CommitQC]()), - } -} - -func (cv *commitVotes) pushVote(c *types.Committee, vote *types.Signed[*types.CommitVote]) { - key := vote.Key() - view := vote.Msg().Proposal().View() - - // Check if the key has already voted. - if oldVote, exists := cv.byKey[key]; exists { - oldView := oldVote.Msg().Proposal().View() - if !oldView.Less(view) { - return // Ignore older or equal votes. - } - // Remove the old vote from the view map. - h := oldVote.Hash() - cv.byHash[h].weight -= c.Weight(key) - delete(cv.byHash[h].votes, key) - if len(cv.byHash[h].votes) == 0 { - delete(cv.byHash, h) - } - } - - // Insert the new vote. - cv.byKey[key] = vote - h := vote.Hash() - if _, exists := cv.byHash[h]; !exists { - cv.byHash[h] = newVoteSet[scv]() - } - cv.byHash[h].weight += c.Weight(key) - cv.byHash[h].votes[key] = vote - - // Check if we have enough votes for a CommitQC. - if cv.byHash[h].weight < c.CommitQuorum() { - return - } - - // Construct a CommitQC from the votes. - old := cv.qc.Load() - if old, ok := old.Get(); ok && !old.Proposal().View().Less(view) { - return - } - var votes []*types.Signed[*types.CommitVote] - for _, v := range cv.byHash[h].votes { - votes = append(votes, v) - } - cv.qc.Store(utils.Some(types.NewCommitQC(votes))) -} diff --git a/sei-tendermint/internal/autobahn/consensus/phase_votes.go b/sei-tendermint/internal/autobahn/consensus/phase_votes.go new file mode 100644 index 0000000000..42af86e0a3 --- /dev/null +++ b/sei-tendermint/internal/autobahn/consensus/phase_votes.go @@ -0,0 +1,92 @@ +package consensus + +import ( + "github.com/sei-protocol/sei-chain/sei-tendermint/autobahn/types" + "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils" +) + +type spv = *types.Signed[*types.PrepareVote] +type scv = *types.Signed[*types.CommitVote] +type hpv = types.Hash[*types.PrepareVote] +type hcv = types.Hash[*types.CommitVote] + +type prepareVotes = phaseVotes[spv, hpv, *types.PrepareQC] +type commitVotes = phaseVotes[scv, hcv, *types.CommitQC] +type timeoutVotes = phaseVotes[*types.FullTimeoutVote, types.View, *types.TimeoutQC] + +// votePhase is the wiring of one consensus phase: how a vote of that phase yields the +// aggregator's inputs, and how a quorum of such votes becomes that phase's QC. +type votePhase[V any, B comparable, QC any] struct { + key func(V) types.PublicKey + view func(V) types.View + bucket func(V) B + quorum func(*types.Committee) uint64 + qcView func(QC) types.View + newQC func([]V) QC +} + +// phaseVotes holds the votes of one consensus phase and publishes the QC they form. +type phaseVotes[V any, B comparable, QC any] struct { + phase votePhase[V, B, QC] + votes *voteAggregator[V, B] + qc utils.AtomicSend[utils.Option[QC]] +} + +func newPhaseVotes[V any, B comparable, QC any](phase votePhase[V, B, QC]) *phaseVotes[V, B, QC] { + return &phaseVotes[V, B, QC]{ + phase: phase, + votes: newVoteAggregator[V, B](), + qc: utils.NewAtomicSend(utils.None[QC]()), + } +} + +// pushVerifiedVote inserts a vote the caller has already verified against c, publishing a QC +// once the vote completes a quorum at a view later than the last QC published. +func (p *phaseVotes[V, B, QC]) pushVerifiedVote(c *types.Committee, vote V) { + ph := p.phase + votes, ok := p.votes.pushVote(c, ph.key(vote), ph.view(vote), ph.bucket(vote), vote, ph.quorum(c)).Get() + if !ok { + return + } + // Construct a QC from the votes. + if old, ok := p.qc.Load().Get(); ok && !ph.qcView(old).Less(ph.view(vote)) { + return + } + p.qc.Store(utils.Some(ph.newQC(votes))) +} + +// newPrepareVotes returns an empty prepare-phase vote aggregator. +func newPrepareVotes() *prepareVotes { + return newPhaseVotes(votePhase[spv, hpv, *types.PrepareQC]{ + key: func(v spv) types.PublicKey { return v.Key() }, + view: func(v spv) types.View { return v.Msg().Proposal().View() }, + bucket: func(v spv) hpv { return v.Hash() }, + quorum: (*types.Committee).PrepareQuorum, + qcView: func(qc *types.PrepareQC) types.View { return qc.Proposal().View() }, + newQC: types.NewPrepareQC, + }) +} + +// newCommitVotes returns an empty commit-phase vote aggregator. +func newCommitVotes() *commitVotes { + return newPhaseVotes(votePhase[scv, hcv, *types.CommitQC]{ + key: func(v scv) types.PublicKey { return v.Key() }, + view: func(v scv) types.View { return v.Msg().Proposal().View() }, + bucket: func(v scv) hcv { return v.Hash() }, + quorum: (*types.Committee).CommitQuorum, + qcView: func(qc *types.CommitQC) types.View { return qc.Proposal().View() }, + newQC: types.NewCommitQC, + }) +} + +// newTimeoutVotes returns an empty timeout-phase vote aggregator. +func newTimeoutVotes() *timeoutVotes { + return newPhaseVotes(votePhase[*types.FullTimeoutVote, types.View, *types.TimeoutQC]{ + key: func(v *types.FullTimeoutVote) types.PublicKey { return v.Vote().Key() }, + view: (*types.FullTimeoutVote).View, + bucket: (*types.FullTimeoutVote).View, + quorum: (*types.Committee).TimeoutQuorum, + qcView: (*types.TimeoutQC).View, + newQC: types.NewTimeoutQC, + }) +} diff --git a/sei-tendermint/internal/autobahn/consensus/phase_votes_test.go b/sei-tendermint/internal/autobahn/consensus/phase_votes_test.go new file mode 100644 index 0000000000..c133300571 --- /dev/null +++ b/sei-tendermint/internal/autobahn/consensus/phase_votes_test.go @@ -0,0 +1,158 @@ +package consensus + +import ( + "testing" + + "github.com/sei-protocol/sei-chain/sei-tendermint/autobahn/types" + "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils" + "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/require" +) + +func TestPrepareVotes_QuorumFormsQC(t *testing.T) { + rng := utils.TestRng() + e := newVoteTestEnv(rng) + pv := newPrepareVotes() + proposal := types.GenProposalForEpoch(rng, e.ep, e.view) + + for _, k := range e.quorum { + pv.pushVerifiedVote(e.ep.Committee(), types.Sign(k, types.NewPrepareVote(proposal))) + } + got, ok := pv.qc.Load().Get() + require.True(t, ok) + require.Equal(t, e.view, got.Proposal().View()) + require.NoError(t, got.Verify(e.ep)) +} + +func TestPrepareVotes_DoesNotReplaceQCAtSameView(t *testing.T) { + rng := utils.TestRng() + e := newVoteTestEnv(rng) + pv := newPrepareVotes() + view1 := e.view.Next() + proposal := types.GenProposalForEpoch(rng, e.ep, view1) + + for _, k := range e.quorum { + pv.pushVerifiedVote(e.ep.Committee(), types.Sign(k, types.NewPrepareVote(proposal))) + } + before, ok := pv.qc.Load().Get() + require.True(t, ok) + require.Equal(t, view1, before.Proposal().View()) + pv.pushVerifiedVote(e.ep.Committee(), types.Sign(e.keys[len(e.quorum)], types.NewPrepareVote(proposal))) + after, ok := pv.qc.Load().Get() + require.True(t, ok) + require.True(t, before == after) +} + +func TestPrepareVotes_ReplacesQCAtNewerView(t *testing.T) { + rng := utils.TestRng() + e := newVoteTestEnv(rng) + pv := newPrepareVotes() + view1 := e.view.Next() + c := e.ep.Committee() + + p0 := types.GenProposalForEpoch(rng, e.ep, e.view) + for _, k := range e.quorum { + pv.pushVerifiedVote(c, types.Sign(k, types.NewPrepareVote(p0))) + } + got, ok := pv.qc.Load().Get() + require.True(t, ok) + require.Equal(t, e.view, got.Proposal().View()) + + p1 := types.GenProposalForEpoch(rng, e.ep, view1) + for _, k := range e.quorum { + pv.pushVerifiedVote(c, types.Sign(k, types.NewPrepareVote(p1))) + } + got, ok = pv.qc.Load().Get() + require.True(t, ok) + require.Equal(t, view1, got.Proposal().View()) + require.NoError(t, got.Verify(e.ep)) +} + +// Prepare votes are bucketed by vote hash, so votes for conflicting proposals at the +// same view never combine into a QC. +func TestPrepareVotes_ConflictingProposalsDoNotFormQC(t *testing.T) { + rng := utils.TestRng() + e := newVoteTestEnv(rng) + pv := newPrepareVotes() + first, second := e.splitBelowQuorum(t) + a := types.GenProposalForEpoch(rng, e.ep, e.view) + b := types.GenProposalForEpoch(rng, e.ep, e.view) + + for _, k := range first { + pv.pushVerifiedVote(e.ep.Committee(), types.Sign(k, types.NewPrepareVote(a))) + } + for _, k := range second { + pv.pushVerifiedVote(e.ep.Committee(), types.Sign(k, types.NewPrepareVote(b))) + } + require.False(t, pv.qc.Load().IsPresent()) +} + +func TestCommitVotes_QuorumFormsQC(t *testing.T) { + rng := utils.TestRng() + e := newVoteTestEnv(rng) + cv := newCommitVotes() + proposal := types.GenProposalForEpoch(rng, e.ep, e.view) + + for _, k := range e.quorum { + cv.pushVerifiedVote(e.ep.Committee(), types.Sign(k, types.NewCommitVote(proposal))) + } + got, ok := cv.qc.Load().Get() + require.True(t, ok) + require.Equal(t, e.view, got.Proposal().View()) + require.NoError(t, got.Verify(e.ep)) +} + +// Commit votes are bucketed by vote hash, so votes for conflicting proposals at the same +// view never combine into a QC. +func TestCommitVotes_ConflictingProposalsDoNotFormQC(t *testing.T) { + rng := utils.TestRng() + e := newVoteTestEnv(rng) + cv := newCommitVotes() + first, second := e.splitBelowQuorum(t) + a := types.GenProposalForEpoch(rng, e.ep, e.view) + b := types.GenProposalForEpoch(rng, e.ep, e.view) + + for _, k := range first { + cv.pushVerifiedVote(e.ep.Committee(), types.Sign(k, types.NewCommitVote(a))) + } + for _, k := range second { + cv.pushVerifiedVote(e.ep.Committee(), types.Sign(k, types.NewCommitVote(b))) + } + require.False(t, cv.qc.Load().IsPresent()) +} + +func TestTimeoutVotes_QuorumFormsQC(t *testing.T) { + rng := utils.TestRng() + e := newVoteTestEnv(rng) + tv := newTimeoutVotes() + pqc := makePrepareQC(e.keys, types.GenProposalForEpoch(rng, e.ep, e.view)) + + for _, k := range e.quorum { + tv.pushVerifiedVote(e.ep.Committee(), types.NewFullTimeoutVote(k, e.view, utils.Some(pqc))) + } + got, ok := tv.qc.Load().Get() + require.True(t, ok) + require.Equal(t, e.view, got.View()) + require.NoError(t, got.Verify(e.ep, utils.None[*types.CommitQC]())) + require.True(t, got.LatestPrepareQC().IsPresent()) +} + +// Timeout votes are bucketed by view rather than by content, so a quorum still forms when +// signers report different prepare QCs. +func TestTimeoutVotes_DifferingPrepareQCsFormQC(t *testing.T) { + rng := utils.TestRng() + e := newVoteTestEnv(rng) + tv := newTimeoutVotes() + pqc := makePrepareQC(e.keys, types.GenProposalForEpoch(rng, e.ep, e.view)) + + for i, k := range e.quorum { + latest := utils.None[*types.PrepareQC]() + if i == 0 { + latest = utils.Some(pqc) + } + tv.pushVerifiedVote(e.ep.Committee(), types.NewFullTimeoutVote(k, e.view, latest)) + } + got, ok := tv.qc.Load().Get() + require.True(t, ok) + require.NoError(t, got.Verify(e.ep, utils.None[*types.CommitQC]())) + require.True(t, got.LatestPrepareQC().IsPresent()) +} diff --git a/sei-tendermint/internal/autobahn/consensus/prepare_votes.go b/sei-tendermint/internal/autobahn/consensus/prepare_votes.go deleted file mode 100644 index 532a34ea29..0000000000 --- a/sei-tendermint/internal/autobahn/consensus/prepare_votes.go +++ /dev/null @@ -1,82 +0,0 @@ -package consensus - -import ( - "github.com/sei-protocol/sei-chain/sei-tendermint/autobahn/types" - "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils" -) - -type spv = *types.Signed[*types.PrepareVote] -type hpv = types.Hash[*types.PrepareVote] - -type voteSet[V any] struct { - weight uint64 - votes map[types.PublicKey]V -} - -func newVoteSet[V any]() *voteSet[V] { - return &voteSet[V]{ - weight: 0, - votes: map[types.PublicKey]V{}, - } -} - -// prepareVotes holds the votes for the prepare phase of consensus. -type prepareVotes struct { - byKey map[types.PublicKey]spv - byHash map[hpv]*voteSet[spv] - qc utils.AtomicSend[utils.Option[*types.PrepareQC]] -} - -// newPrepareVotes initializes a new prepareVotes instance. -func newPrepareVotes() *prepareVotes { - return &prepareVotes{ - byKey: map[types.PublicKey]spv{}, - byHash: map[hpv]*voteSet[spv]{}, - qc: utils.NewAtomicSend(utils.None[*types.PrepareQC]()), - } -} - -// pushVote processes a new prepare vote and updates the prepare votes state. -func (pv *prepareVotes) pushVote(c *types.Committee, vote *types.Signed[*types.PrepareVote]) { - key := vote.Key() - view := vote.Msg().Proposal().View() - - // Check if the key has already voted. - if oldVote, exists := pv.byKey[key]; exists { - oldView := oldVote.Msg().Proposal().View() - if !oldView.Less(view) { - return // Ignore older or equal votes. - } - // Remove the old vote from the view map. - h := oldVote.Hash() - pv.byHash[h].weight -= c.Weight(key) - delete(pv.byHash[h].votes, key) - if len(pv.byHash[h].votes) == 0 { - delete(pv.byHash, h) - } - } - - // Insert the new vote. - pv.byKey[key] = vote - h := vote.Hash() - if _, exists := pv.byHash[h]; !exists { - pv.byHash[h] = newVoteSet[spv]() - } - pv.byHash[h].weight += c.Weight(key) - pv.byHash[h].votes[key] = vote - - // Check if we have enough votes for a PrepareQC. - if pv.byHash[h].weight < c.PrepareQuorum() { - return - } - - // Construct a PrepareQC from the votes. - if old, ok := pv.qc.Load().Get(); ok && !old.Proposal().View().Less(view) { - return - } - var votes []*types.Signed[*types.PrepareVote] - for _, v := range pv.byHash[h].votes { - votes = append(votes, v) - } - pv.qc.Store(utils.Some(types.NewPrepareQC(votes))) -} diff --git a/sei-tendermint/internal/autobahn/consensus/state.go b/sei-tendermint/internal/autobahn/consensus/state.go index f3d47ae2f8..b606a96d1a 100644 --- a/sei-tendermint/internal/autobahn/consensus/state.go +++ b/sei-tendermint/internal/autobahn/consensus/state.go @@ -194,7 +194,7 @@ func (s *State) PushPrepareVote(vote *types.Signed[*types.PrepareVote]) error { return fmt.Errorf("vote.VerifySig(): %w", err) } for pv := range s.prepareVotes.Lock() { - pv.pushVote(committee, vote) + pv.pushVerifiedVote(committee, vote) } return nil } @@ -209,7 +209,7 @@ func (s *State) PushCommitVote(vote *types.Signed[*types.CommitVote]) error { return fmt.Errorf("vote.VerifySig(): %w", err) } for cv := range s.commitVotes.Lock() { - cv.pushVote(committee, vote) + cv.pushVerifiedVote(committee, vote) } return nil } diff --git a/sei-tendermint/internal/autobahn/consensus/timeout_votes.go b/sei-tendermint/internal/autobahn/consensus/timeout_votes.go deleted file mode 100644 index bfa3771262..0000000000 --- a/sei-tendermint/internal/autobahn/consensus/timeout_votes.go +++ /dev/null @@ -1,60 +0,0 @@ -package consensus - -import ( - "github.com/sei-protocol/sei-chain/sei-tendermint/autobahn/types" - "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils" -) - -type timeoutVotes struct { - byKey map[types.PublicKey]*types.FullTimeoutVote - byView map[types.View]*voteSet[*types.FullTimeoutVote] - qc utils.AtomicSend[utils.Option[*types.TimeoutQC]] -} - -func newTimeoutVotes() *timeoutVotes { - return &timeoutVotes{ - byKey: map[types.PublicKey]*types.FullTimeoutVote{}, - byView: map[types.View]*voteSet[*types.FullTimeoutVote]{}, - qc: utils.NewAtomicSend(utils.None[*types.TimeoutQC]()), - } -} - -// pushVerifiedVote inserts a timeout vote the caller has already verified against c. -func (tv *timeoutVotes) pushVerifiedVote(c *types.Committee, vote *types.FullTimeoutVote) { - key := vote.Vote().Key() - view := vote.Vote().Msg().View() - if old, ok := tv.byKey[key]; ok { - // Check if the old vote is newer than the new one. - oldView := old.Vote().Msg().View() - if !oldView.Less(view) { - return - } - // Prune the old vote. - tv.byView[oldView].weight -= c.Weight(key) - delete(tv.byView[oldView].votes, key) - if len(tv.byView[oldView].votes) == 0 { - delete(tv.byView, oldView) - } - } - // Insert the new vote. - tv.byKey[key] = vote - if _, ok := tv.byView[view]; !ok { - tv.byView[view] = newVoteSet[*types.FullTimeoutVote]() - } - tv.byView[view].weight += c.Weight(key) - tv.byView[view].votes[key] = vote - // Check if we have enough votes for a TimeoutQC. - if tv.byView[view].weight < c.TimeoutQuorum() { - return - } - // Construct a TimeoutQC from the votes. - old := tv.qc.Load() - if old, ok := old.Get(); ok && !old.View().Less(view) { - return - } - var votes []*types.FullTimeoutVote - for _, v := range tv.byView[view].votes { - votes = append(votes, v) - } - tv.qc.Store(utils.Some(types.NewTimeoutQC(votes))) -} diff --git a/sei-tendermint/internal/autobahn/consensus/timeout_votes_test.go b/sei-tendermint/internal/autobahn/consensus/timeout_votes_test.go deleted file mode 100644 index b70cc2cbba..0000000000 --- a/sei-tendermint/internal/autobahn/consensus/timeout_votes_test.go +++ /dev/null @@ -1,120 +0,0 @@ -package consensus - -import ( - "testing" - "time" - - "github.com/sei-protocol/sei-chain/sei-tendermint/autobahn/types" - "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils" - "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/require" -) - -// timeoutEnv is a timeoutVotes aggregator over a 4-replica equal-weight committee. -type timeoutEnv struct { - tv *timeoutVotes - keys []types.SecretKey - ep *types.Epoch - view types.View - quorum []types.SecretKey -} - -func newTimeoutEnv(rng utils.Rng) timeoutEnv { - keys := utils.GenSliceN(rng, 4, types.GenSecretKey) - weights := make(map[types.PublicKey]uint64, len(keys)) - for _, k := range keys { - weights[k.Public()] = 1 - } - c := utils.OrPanic1(types.NewCommittee(weights)) - ep := types.NewEpoch(0, types.OpenRoadRange(), time.Time{}, c, 0) - view := types.View{Index: 0, Number: 0, EpochIndex: ep.EpochIndex()} - quorum := types.TestKeysWithWeight(c, keys, c.TimeoutQuorum()) - return timeoutEnv{ - tv: newTimeoutVotes(), - keys: keys, - ep: ep, - view: view, - quorum: quorum, - } -} - -func TestTimeoutVotes_BelowQuorumDoesNotFormQC(t *testing.T) { - e := newTimeoutEnv(utils.TestRng()) - c := e.ep.Committee() - - e.tv.pushVerifiedVote(c, types.NewFullTimeoutVote(e.keys[0], e.view, utils.None[*types.PrepareQC]())) - require.False(t, e.tv.qc.Load().IsPresent()) -} - -func TestTimeoutVotes_QuorumFormsQC(t *testing.T) { - rng := utils.TestRng() - e := newTimeoutEnv(rng) - c := e.ep.Committee() - pqc := makePrepareQC(e.keys, types.GenProposalForEpoch(rng, e.ep, e.view)) - - for _, k := range e.quorum { - e.tv.pushVerifiedVote(c, types.NewFullTimeoutVote(k, e.view, utils.Some(pqc))) - } - got, ok := e.tv.qc.Load().Get() - require.True(t, ok) - require.Equal(t, e.view, got.View()) - require.NoError(t, got.Verify(e.ep, utils.None[*types.CommitQC]())) - require.True(t, got.LatestPrepareQC().IsPresent()) -} - -func TestTimeoutVotes_SameViewAfterQCDoesNotChangeView(t *testing.T) { - e := newTimeoutEnv(utils.TestRng()) - c := e.ep.Committee() - - for _, k := range e.quorum { - e.tv.pushVerifiedVote(c, types.NewFullTimeoutVote(k, e.view, utils.None[*types.PrepareQC]())) - } - before, ok := e.tv.qc.Load().Get() - require.True(t, ok) - e.tv.pushVerifiedVote(c, types.NewFullTimeoutVote(e.keys[len(e.quorum)], e.view, utils.None[*types.PrepareQC]())) - got, ok := e.tv.qc.Load().Get() - require.True(t, ok) - require.True(t, before == got) -} - -func TestTimeoutVotes_IgnoresStaleVoteFromSameKey(t *testing.T) { - e := newTimeoutEnv(utils.TestRng()) - c := e.ep.Committee() - view1 := e.view.Next() - - for _, k := range e.quorum { - e.tv.pushVerifiedVote(c, types.NewFullTimeoutVote(k, view1, utils.None[*types.PrepareQC]())) - } - got, ok := e.tv.qc.Load().Get() - require.True(t, ok) - require.Equal(t, view1, got.View()) - - e.tv.pushVerifiedVote(c, types.NewFullTimeoutVote(e.quorum[0], e.view, utils.None[*types.PrepareQC]())) - got, ok = e.tv.qc.Load().Get() - require.True(t, ok) - require.Equal(t, view1, got.View()) -} - -func TestTimeoutVotes_ReplacesOlderVoteAndAdvancesQC(t *testing.T) { - e := newTimeoutEnv(utils.TestRng()) - c := e.ep.Committee() - view1 := e.view.Next() - - for _, k := range e.quorum { - e.tv.pushVerifiedVote(c, types.NewFullTimeoutVote(k, e.view, utils.None[*types.PrepareQC]())) - } - got, ok := e.tv.qc.Load().Get() - require.True(t, ok) - require.Equal(t, e.view, got.View()) - - e.tv.pushVerifiedVote(c, types.NewFullTimeoutVote(e.quorum[0], view1, utils.None[*types.PrepareQC]())) - got, ok = e.tv.qc.Load().Get() - require.True(t, ok) - require.Equal(t, e.view, got.View()) - - for _, k := range e.quorum[1:] { - e.tv.pushVerifiedVote(c, types.NewFullTimeoutVote(k, view1, utils.None[*types.PrepareQC]())) - } - got, ok = e.tv.qc.Load().Get() - require.True(t, ok) - require.Equal(t, view1, got.View()) -} diff --git a/sei-tendermint/internal/autobahn/consensus/vote_aggregator.go b/sei-tendermint/internal/autobahn/consensus/vote_aggregator.go new file mode 100644 index 0000000000..0603298271 --- /dev/null +++ b/sei-tendermint/internal/autobahn/consensus/vote_aggregator.go @@ -0,0 +1,72 @@ +package consensus + +import ( + "github.com/sei-protocol/sei-chain/sei-tendermint/autobahn/types" + "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils" +) + +type voteSet[V any] struct { + weight uint64 + votes map[types.PublicKey]V +} + +type voteEntry[V any, B comparable] struct { + vote V + view types.View + bucket B +} + +type voteAggregator[V any, B comparable] struct { + byKey map[types.PublicKey]voteEntry[V, B] + byBucket map[B]*voteSet[V] +} + +func newVoteAggregator[V any, B comparable]() *voteAggregator[V, B] { + return &voteAggregator[V, B]{ + byKey: map[types.PublicKey]voteEntry[V, B]{}, + byBucket: map[B]*voteSet[V]{}, + } +} + +func (a *voteAggregator[V, B]) pushVote( + c *types.Committee, + key types.PublicKey, + view types.View, + bucket B, + vote V, + quorum uint64, +) utils.Option[[]V] { + // Check if the key has already voted. + if old, ok := a.byKey[key]; ok { + if !old.view.Less(view) { + return utils.None[[]V]() // Ignore older or equal votes. + } + // Prune the old vote. + oldSet := a.byBucket[old.bucket] + oldSet.weight -= c.Weight(key) + delete(oldSet.votes, key) + if len(oldSet.votes) == 0 { + delete(a.byBucket, old.bucket) + } + } + + // Insert the new vote. + a.byKey[key] = voteEntry[V, B]{vote: vote, view: view, bucket: bucket} + if _, ok := a.byBucket[bucket]; !ok { + a.byBucket[bucket] = &voteSet[V]{votes: map[types.PublicKey]V{}} + } + set := a.byBucket[bucket] + set.weight += c.Weight(key) + set.votes[key] = vote + + // Check if we have enough votes for a QC. + if set.weight < quorum { + return utils.None[[]V]() + } + + votes := make([]V, 0, len(set.votes)) + for _, vote := range set.votes { + votes = append(votes, vote) + } + return utils.Some(votes) +} diff --git a/sei-tendermint/internal/autobahn/consensus/vote_aggregator_test.go b/sei-tendermint/internal/autobahn/consensus/vote_aggregator_test.go new file mode 100644 index 0000000000..86b62c4162 --- /dev/null +++ b/sei-tendermint/internal/autobahn/consensus/vote_aggregator_test.go @@ -0,0 +1,143 @@ +package consensus + +import ( + "testing" + "time" + + "github.com/sei-protocol/sei-chain/sei-tendermint/autobahn/types" + "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils" + "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/require" +) + +type voteTestEnv struct { + keys []types.SecretKey + ep *types.Epoch + view types.View + quorum []types.SecretKey +} + +func newVoteTestEnv(rng utils.Rng) voteTestEnv { + keys := utils.GenSliceN(rng, 4, types.GenSecretKey) + weights := make(map[types.PublicKey]uint64, len(keys)) + for _, k := range keys { + weights[k.Public()] = 1 + } + c := utils.OrPanic1(types.NewCommittee(weights)) + ep := types.NewEpoch(0, types.OpenRoadRange(), time.Time{}, c, 0) + return voteTestEnv{ + keys: keys, + ep: ep, + view: types.View{Index: 0, Number: 0, EpochIndex: ep.EpochIndex()}, + quorum: types.TestKeysWithWeight(c, keys, c.CommitQuorum()), + } +} + +// keyWeight is the total committee weight held by keys. +func keyWeight(c *types.Committee, keys []types.SecretKey) uint64 { + total := uint64(0) + for _, k := range keys { + total += c.Weight(k.Public()) + } + return total +} + +// splitBelowQuorum divides the committee into two groups that each fall short of quorum +// but together exceed it, so a QC forming across both groups proves their votes shared a +// bucket rather than reaching quorum on their own. +func (e voteTestEnv) splitBelowQuorum(t *testing.T) (first, second []types.SecretKey) { + c := e.ep.Committee() + first, second = e.keys[:len(e.keys)/2], e.keys[len(e.keys)/2:] + require.True(t, keyWeight(c, first) < c.CommitQuorum()) + require.True(t, keyWeight(c, second) < c.CommitQuorum()) + require.True(t, keyWeight(c, first)+keyWeight(c, second) >= c.CommitQuorum()) + return first, second +} + +type testAggregatedVote struct { + key types.PublicKey + view types.View + bucket int +} + +type aggregationTestEnv struct { + voteTestEnv + votes *voteAggregator[*testAggregatedVote, int] +} + +func newAggregationTestEnv(rng utils.Rng) aggregationTestEnv { + return aggregationTestEnv{ + voteTestEnv: newVoteTestEnv(rng), + votes: newVoteAggregator[*testAggregatedVote, int](), + } +} + +func (e aggregationTestEnv) push(key types.SecretKey, view types.View, bucket int) utils.Option[[]*testAggregatedVote] { + vote := &testAggregatedVote{key: key.Public(), view: view, bucket: bucket} + return e.votes.pushVote(e.ep.Committee(), vote.key, vote.view, vote.bucket, vote, e.ep.Committee().CommitQuorum()) +} + +func TestVoteAggregator_BelowQuorumDoesNotEmit(t *testing.T) { + e := newAggregationTestEnv(utils.TestRng()) + + for _, k := range e.quorum[:len(e.quorum)-1] { + require.False(t, e.push(k, e.view, 0).IsPresent()) + } +} + +func TestVoteAggregator_QuorumEmitsVotes(t *testing.T) { + e := newAggregationTestEnv(utils.TestRng()) + var got utils.Option[[]*testAggregatedVote] + + for _, k := range e.quorum { + got = e.push(k, e.view, 0) + } + votes, ok := got.Get() + require.True(t, ok) + require.Len(t, votes, len(e.quorum)) +} + +func TestVoteAggregator_IgnoresEqualVoteFromSameKey(t *testing.T) { + e := newAggregationTestEnv(utils.TestRng()) + + e.push(e.quorum[0], e.view, 0) + e.push(e.quorum[0], e.view, 1) + var got utils.Option[[]*testAggregatedVote] + for _, k := range e.quorum[1:] { + got = e.push(k, e.view, 0) + } + votes, ok := got.Get() + require.True(t, ok) + require.Len(t, votes, len(e.quorum)) +} + +func TestVoteAggregator_IgnoresStaleVoteFromSameKey(t *testing.T) { + e := newAggregationTestEnv(utils.TestRng()) + view1 := e.view.Next() + + e.push(e.quorum[0], view1, 1) + e.push(e.quorum[0], e.view, 0) + var got utils.Option[[]*testAggregatedVote] + for _, k := range e.quorum[1:] { + got = e.push(k, view1, 1) + } + votes, ok := got.Get() + require.True(t, ok) + require.Len(t, votes, len(e.quorum)) +} + +func TestVoteAggregator_ReplacesOlderVotesAndEmitsAtNewView(t *testing.T) { + e := newAggregationTestEnv(utils.TestRng()) + view1 := e.view.Next() + + for _, k := range e.quorum { + e.push(k, e.view, 0) + } + require.False(t, e.push(e.quorum[0], view1, 1).IsPresent()) + var got utils.Option[[]*testAggregatedVote] + for _, k := range e.quorum[1:] { + got = e.push(k, view1, 1) + } + votes, ok := got.Get() + require.True(t, ok) + require.Len(t, votes, len(e.quorum)) +}