(Autobahn) Share vote aggregation (CON-423)

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,
+	})
+}

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())
+}