chore(halo/voter): improve errors and logs

halo/app/lazyvoter.go

@@ -166,10 +166,10 @@ func (l *voterLoader) LazyLoad(
 	defer l.mu.Unlock()
 
 	// Process all cached values
-	if err := v.SetProposed(l.proposed); err != nil {
+	if err := v.SetProposed(ctx, l.proposed); err != nil {
 		return errors.Wrap(err, "set cached proposed")
 	}
-	if err := v.SetCommitted(l.committed); err != nil {
+	if err := v.SetCommitted(ctx, l.committed); err != nil {
 		return errors.Wrap(err, "set cached committed")
 	}
 	if l.lastValSet != nil {
@@ -212,9 +212,9 @@ func (l *voterLoader) GetAvailable() []*atypes.Vote {
 	return nil // Return empty list if voter not available yet.
 }
 
-func (l *voterLoader) SetProposed(headers []*atypes.AttestHeader) error {
+func (l *voterLoader) SetProposed(ctx context.Context, headers []*atypes.AttestHeader) error {
 	if v, ok := l.getVoter(); ok {
-		return v.SetProposed(headers)
+		return v.SetProposed(ctx, headers)
 	}
 
 	// Cache these headers to provider to voter once available.
@@ -226,9 +226,9 @@ func (l *voterLoader) SetProposed(headers []*atypes.AttestHeader) error {
 	return nil
 }
 
-func (l *voterLoader) SetCommitted(headers []*atypes.AttestHeader) error {
+func (l *voterLoader) SetCommitted(ctx context.Context, headers []*atypes.AttestHeader) error {
 	if v, ok := l.getVoter(); ok {
-		return v.SetCommitted(headers)
+		return v.SetCommitted(ctx, headers)
 	}
 
 	// Cache these headers to provider to voter once available.

halo/attest/keeper/keeper.go

@@ -3,9 +3,6 @@ package keeper
 import (
 	"bytes"
 	"context"
-	"fmt"
-	"log/slog"
-	"strconv"
 
 	"github.com/omni-network/omni/halo/attest/types"
 	rtypes "github.com/omni-network/omni/halo/registry/types"
@@ -713,29 +710,7 @@ func (k *Keeper) ExtendVote(ctx sdk.Context, _ *abci.RequestExtendVote) (*abci.R
 		votesExtended.WithLabelValues(k.namer(chainVer)).Observe(float64(count))
 	}
 
-	// Make nice logs
-	const limit = 5
-	offsets := make(map[xchain.ChainVersion][]string)
-	for _, vote := range filtered {
-		offset := offsets[vote.AttestHeader.XChainVersion()]
-		if len(offset) < limit {
-			offset = append(offset, strconv.FormatUint(vote.AttestHeader.AttestOffset, 10))
-		} else if len(offset) == limit {
-			offset = append(offset, "...")
-		} else {
-			continue
-		}
-		offsets[vote.AttestHeader.XChainVersion()] = offset
-	}
-	attrs := []any{slog.Int("votes", len(offsets))}
-	for chainVer, offset := range offsets {
-		attrs = append(attrs, slog.String(
-			fmt.Sprintf("%d-%d", chainVer.ID, chainVer.ConfLevel),
-			fmt.Sprint(offset),
-		))
-	}
-
-	log.Info(ctx, "Voted for rollup blocks", attrs...)
+	log.Info(ctx, "Voter voting", types.VoteLogs(filtered)...)
 
 	return &abci.ResponseExtendVote{
 		VoteExtension: bz,
@@ -780,20 +755,31 @@ func (k *Keeper) VerifyVoteExtension(ctx sdk.Context, req *abci.RequestVerifyVot
 		return respReject, nil
 	}
 
-	duplicate := make(map[xchain.AttestHeader]bool)
+	duplicate := make(map[common.Hash]bool)          // Detect identical duplicate votes (same AttestationRoot)
+	doubleSign := make(map[xchain.AttestHeader]bool) // Detect double sign votes (same AttestHeader)
 	for _, vote := range votes.Votes {
 		if err := vote.Verify(); err != nil {
 			log.Warn(ctx, "Rejecting invalid vote", err)
 			return respReject, nil
 		}
 
-		if duplicate[vote.AttestHeader.ToXChain()] {
+		attRoot, err := vote.AttestationRoot()
+		if err != nil {
+			return nil, errors.Wrap(err, "att root [BUG]") // Should error in Verify
+		}
+		if duplicate[attRoot] {
+			log.Warn(ctx, "Rejecting duplicate identical vote", nil)
+			return respReject, nil
+		}
+		duplicate[attRoot] = true
+
+		if doubleSign[vote.AttestHeader.ToXChain()] {
 			doubleSignCounter.WithLabelValues(ethAddr.Hex()).Inc()
 			log.Warn(ctx, "Rejecting duplicate slashable vote", err)
 
 			return respReject, nil
 		}
-		duplicate[vote.AttestHeader.ToXChain()] = true
+		doubleSign[vote.AttestHeader.ToXChain()] = true
 
 		// Ensure the votes are from the requesting validator itself.
 		if !bytes.Equal(vote.Signature.ValidatorAddress, ethAddr[:]) {
@@ -880,6 +866,7 @@ func (k *Keeper) windowCompare(ctx context.Context, chainVer xchain.ChainVersion
 // - Ensure all aggregation is valid; no duplicate aggregate votes.
 // - Ensure the vote extension limit is not exceeded per validator.
 // - Ensure all votes are from validators in the provided set.
+// - Ensure all votes are unique per validator.
 // - Ensure the vote block header is in the vote window.
 func (k *Keeper) verifyAggVotes(
 	ctx context.Context,
@@ -911,12 +898,18 @@ func (k *Keeper) verifyAggVotes(
 		duplicate[attRoot] = true
 
 		// Ensure all votes are from unique validators in the set
+		duplicateSig := make(map[common.Address]bool) // Enforce vote extension limit.
 		for _, sig := range agg.Signatures {
 			addr, err := sig.ValidatorEthAddress()
 			if err != nil {
 				return err
 			}
 
+			if duplicateSig[addr] {
+				return errors.New("duplicate validator vote", append(errAttrs, "validator", addr)...)
+			}
+			duplicateSig[addr] = true
+
 			if !valset.Contains(addr) {
 				return errors.New("vote from unknown validator", append(errAttrs, "validator", addr)...)
 			}

halo/attest/keeper/msg_server.go

@@ -3,13 +3,9 @@ package keeper
 import (
 	"bytes"
 	"context"
-	"fmt"
-	"log/slog"
-	"strconv"
 
 	"github.com/omni-network/omni/halo/attest/types"
 	"github.com/omni-network/omni/lib/errors"
-	"github.com/omni-network/omni/lib/log"
 
 	"github.com/ethereum/go-ethereum/common"
 
@@ -43,8 +39,7 @@ func (s msgServer) AddVotes(ctx context.Context, msg *types.MsgAddVotes,
 
 	// Update the voter state with the local headers.
 	localHeaders := headersByAddress(msg.Votes, s.voter.LocalAddress())
-	logLocalVotes(ctx, localHeaders, "committed")
-	if err := s.voter.SetCommitted(localHeaders); err != nil {
+	if err := s.voter.SetCommitted(ctx, localHeaders); err != nil {
 		return nil, errors.Wrap(err, "set committed")
 	}
 
@@ -73,34 +68,3 @@ func headersByAddress(aggregates []*types.AggVote, address common.Address) []*ty
 
 	return filtered
 }
-
-func logLocalVotes(ctx context.Context, headers []*types.AttestHeader, typ string) {
-	if len(headers) == 0 {
-		return
-	}
-
-	const limit = 5
-	offsets := make(map[uint64][]string)
-	for _, header := range headers {
-		offset := offsets[header.SourceChainId]
-		if len(offset) == limit {
-			offset = append(offset, "...")
-		} else if len(offset) < limit {
-			offset = append(offset, strconv.FormatUint(header.AttestOffset, 10))
-		} else {
-			continue
-		}
-		offsets[header.SourceChainId] = offset
-	}
-	attrs := []any{
-		slog.Int("votes", len(headers)),
-	}
-	for cid, hs := range offsets {
-		attrs = append(attrs, slog.String(
-			strconv.FormatUint(cid, 10),
-			fmt.Sprint(hs),
-		))
-	}
-
-	log.Debug(ctx, "Marked local votes as "+typ, attrs...)
-}

halo/attest/keeper/proposal_server.go

@@ -33,8 +33,7 @@ func (s proposalServer) AddVotes(ctx context.Context, msg *types.MsgAddVotes,
 	}
 
 	localHeaders := headersByAddress(msg.Votes, s.voter.LocalAddress())
-	logLocalVotes(ctx, localHeaders, "proposed")
-	if err := s.voter.SetProposed(localHeaders); err != nil {
+	if err := s.voter.SetProposed(ctx, localHeaders); err != nil {
 		return nil, errors.Wrap(err, "set committed")
 	}
 

halo/attest/types/helper.go

@@ -0,0 +1,47 @@
+package types
+
+import (
+	"fmt"
+	"log/slog"
+	"strconv"
+
+	"github.com/omni-network/omni/lib/xchain"
+)
+
+const logLimit = 5
+
+// VoteLogs returns the logs as opinionated human-readable logging attributes.
+func VoteLogs(votes []*Vote) []any {
+	var headers []*AttestHeader
+	for _, vote := range votes {
+		headers = append(headers, vote.GetAttestHeader())
+	}
+
+	return AttLogs(headers)
+}
+
+// AttLogs returns the headers as opinionated human-readable logging attributes.
+func AttLogs(headers []*AttestHeader) []any {
+	offsets := make(map[xchain.ChainVersion][]string)
+	for _, header := range headers {
+		offset := offsets[header.XChainVersion()]
+		if len(offset) < logLimit {
+			offset = append(offset, strconv.FormatUint(header.AttestOffset, 10))
+		} else if len(offset) == logLimit {
+			offset = append(offset, "...")
+		} else {
+			continue
+		}
+		offsets[header.XChainVersion()] = offset
+	}
+
+	attrs := []any{slog.Int("count", len(offsets))}
+	for chainVer, offsets := range offsets {
+		attrs = append(attrs, slog.String(
+			fmt.Sprintf("%d-%d", chainVer.ID, chainVer.ConfLevel),
+			fmt.Sprint(offsets),
+		))
+	}
+
+	return attrs
+}

halo/attest/types/interface.go

@@ -24,14 +24,14 @@ type Voter interface {
 	// i.e., they were included by a proposer in a new proposed block.
 	// All other existing "proposed" votes are reset to "available", i.e. they were
 	// proposed previously by another proposer, but that block was never finalized/committed.
-	SetProposed(headers []*AttestHeader) error
+	SetProposed(ctx context.Context, headers []*AttestHeader) error
 
 	// SetCommitted updates the status of the provided votes to "committed",
 	// i.e., they were included in a finalized consensus block and is now part of the consensus chain.
 	// All other existing "proposed" votes are reset to "available", i.e. we probably
 	// missed the proposal step and only learnt of the finalized block post-fact.
 	// All but the latest "confirmed" attestation for each source chain can be safely deleted from disk.
-	SetCommitted(headers []*AttestHeader) error
+	SetCommitted(ctx context.Context, headers []*AttestHeader) error
 
 	// LocalAddress returns the local validator's ethereum address.
 	LocalAddress() common.Address