eth: implement ideal deterministic tx broadcast

Compared to the previous approach, which was probabilistic, the new peer choice will
always select exactly sqrt(peers) for any transaction. After some careful consideration,
it has been determined that use of cryptographic hash function is not required for this
algorithm.

eth/handler.go

@@ -17,20 +17,18 @@
 package eth
 
 import (
-	"bytes"
+	"cmp"
-	"crypto/sha256"
 	"errors"
 	"hash"
+	"hash/fnv"
 	"maps"
 	gmath "math"
-	"math/big"
 	"slices"
 	"sync"
 	"sync/atomic"
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
-	"github.com/ethereum/go-ethereum/common/math"
 	"github.com/ethereum/go-ethereum/core"
 	"github.com/ethereum/go-ethereum/core/rawdb"
 	"github.com/ethereum/go-ethereum/core/txpool"
@@ -485,36 +483,38 @@ func (h *handler) BroadcastTransactions(txs types.Transactions) {
 		annos = make(map[*ethPeer][]common.Hash) // Set peer->hash to announce
 
 		signer = types.LatestSigner(h.chain.Config())
-		choice = newBroadcastChoice(h.peers.len())
+		choice = newBroadcastChoice(h.nodeID)
+		peers  = h.peers.all()
 	)
 
 	for _, tx := range txs {
-		var maybeDirect bool
+		var directSet map[*ethPeer]struct{}
 		switch {
 		case tx.Type() == types.BlobTxType:
 			blobTxs++
 		case tx.Size() > txMaxBroadcastSize:
 			largeTxs++
 		default:
-			maybeDirect = true
+			// Get transaction sender address. Here we can ignore any error
+			// since we're just interested in any value.
+			txSender, _ := types.Sender(signer, tx)
+			directSet = choice.choosePeers(peers, txSender)
 		}
 
-		for _, peer := range h.peers.peersWithoutTransaction(tx.Hash()) {
+		for _, peer := range peers {
-			if maybeDirect {
+			if peer.KnownTransaction(tx.Hash()) {
-				// Get transaction sender address. Here we can ignore any error
+				continue
-				// since we're just interested in any value.
+			}
-				txSender, _ := types.Sender(signer, tx)
+			if _, ok := directSet[peer]; ok {
-				if choice.shouldBroadcastTx(h.nodeID, peer.Peer.Peer.ID(), txSender) {
+				// Send direct.
-					// Send directly to peer.
+				txset[peer] = append(txset[peer], tx.Hash())
-					txset[peer] = append(txset[peer], tx.Hash())
+			} else {
-					continue
+				// Send announcement.
-				}
+				annos[peer] = append(annos[peer], tx.Hash())
 			}
-
-			// Send announcement to peer.
-			annos[peer] = append(annos[peer], tx.Hash())
 		}
 	}
+
 	for peer, hashes := range txset {
 		directCount += len(hashes)
 		peer.AsyncSendTransactions(hashes)
@@ -689,34 +689,49 @@ func (st *blockRangeState) currentRange() eth.BlockRangeUpdatePacket {
 // `sha(self, peer, sender) mod peers < sqrt(peers)`.
 
 type broadcastChoice struct {
-	hash      hash.Hash
+	self   enode.ID
-	threshold []byte
+	hash   hash.Hash64
-	hashBytes []byte
+	buffer map[*ethPeer]struct{}
+	tmp    []broadcastPeer
 }
 
-func newBroadcastChoice(npeers int) *broadcastChoice {
+type broadcastPeer struct {
-	var bc broadcastChoice
+	p     *ethPeer
-	bc.hash = sha256.New()
+	score uint64
-	bc.hashBytes = make([]byte, 32)
-
-	// compute the hash comparison threshold for one peer
-	unit := math.BigPow(2, 256)
-	unit.Sub(unit, common.Big1)
-	unit.Div(unit, big.NewInt(int64(max(1, npeers))))
-
-	// compute the threshold for n peers
-	sqrtp := max(gmath.Sqrt(float64(npeers)), 1)
-	thr := big.NewInt(int64(gmath.Ceil(sqrtp)))
-	thr.Mul(thr, unit)
-	bc.threshold = thr.FillBytes(make([]byte, 32))
-	return &bc
 }
 
-func (bc *broadcastChoice) shouldBroadcastTx(self, peer enode.ID, txSender common.Address) bool {
+func newBroadcastChoice(self enode.ID) *broadcastChoice {
-	bc.hash.Reset()
+	return &broadcastChoice{
-	bc.hash.Write(self[:])
+		self:   self,
-	bc.hash.Write(peer[:])
+		hash:   fnv.New64(),
-	bc.hash.Write(txSender[:])
+		buffer: make(map[*ethPeer]struct{}),
-	bc.hash.Sum(bc.hashBytes[:0])
+	}
-	return bytes.Compare(bc.hashBytes, bc.threshold) < 0
+}
+
+// choosePeers selects the peers that will receive a direct transaction broadcast message.
+//
+// Note the return value will only stay valid until the next call to choosePeers.
+func (bc *broadcastChoice) choosePeers(peers []*ethPeer, txSender common.Address) map[*ethPeer]struct{} {
+	// Compute scores.
+	bc.tmp = bc.tmp[:0]
+	for _, peer := range peers {
+		bc.hash.Reset()
+		bc.hash.Write(bc.self[:])
+		bc.hash.Write(peer.Peer.Peer.ID().Bytes())
+		bc.hash.Write(txSender[:])
+		bc.tmp = append(bc.tmp, broadcastPeer{peer, bc.hash.Sum64()})
+	}
+
+	// Sort by score.
+	slices.SortFunc(bc.tmp, func(a, b broadcastPeer) int {
+		return cmp.Compare(a.score, b.score)
+	})
+
+	// Take top n.
+	clear(bc.buffer)
+	n := int(gmath.Ceil(gmath.Sqrt(float64(len(bc.tmp)))))
+	for i := range n {
+		bc.buffer[bc.tmp[i].p] = struct{}{}
+	}
+	return bc.buffer
 }

eth/handler_test.go

@@ -17,6 +17,7 @@
 package eth
 
 import (
+	"maps"
 	"math/big"
 	"math/rand"
 	"sort"
@@ -31,8 +32,10 @@ import (
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/eth/ethconfig"
+	"github.com/ethereum/go-ethereum/eth/protocols/eth"
 	"github.com/ethereum/go-ethereum/ethdb"
 	"github.com/ethereum/go-ethereum/event"
+	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/p2p/enode"
 	"github.com/ethereum/go-ethereum/params"
 	"github.com/ethereum/go-ethereum/rlp"
@@ -217,64 +220,63 @@ func (b *testHandler) close() {
 }
 
 func TestBroadcastChoice(t *testing.T) {
-	choice49 := newBroadcastChoice(49)
+	// Create choices.
-	choice50 := newBroadcastChoice(50)
+	self := enode.HexID("1111111111111111111111111111111111111111111111111111111111111111")
+	choice49 := newBroadcastChoice(self)
+	choice50 := newBroadcastChoice(self)
 
+	// Create test peers.
 	var (
-		self      = enode.HexID("1111111111111111111111111111111111111111111111111111111111111111")
+		rand  = rand.New(rand.NewSource(33))
-		peers     = make([]enode.ID, 50)
+		peers = make([]*ethPeer, 50)
-		txsenders = make([]common.Address, 400)
-		rand      = rand.New(rand.NewSource(33))
 	)
 	for i := range peers {
-		rand.Read(peers[i][:])
+		var id enode.ID
+		rand.Read(id[:])
+		p2pPeer := p2p.NewPeer(id, "test", nil)
+		ep := eth.NewPeer(eth.ETH69, p2pPeer, nil, nil)
+		peers[i] = &ethPeer{Peer: ep}
 	}
+	defer func() {
+		for _, p := range peers {
+			p.Close()
+		}
+	}()
+
+	// Create random tx sender addresses.
+	txsenders := make([]common.Address, 400)
 	for i := range txsenders {
 		rand.Read(txsenders[i][:])
 	}
 
 	// Evaluate choice49 first.
-	var chosen49 = make([][]bool, len(txsenders))
+	expectedCount := 7 // sqrt(49)
+	var chosen49 = make([]map[*ethPeer]struct{}, len(txsenders))
 	for i, txSender := range txsenders {
-		chosen49[i] = make([]bool, len(peers))
+		set := choice49.choosePeers(peers[:49], txSender)
-		for peerIndex, peer := range peers {
+		chosen49[i] = maps.Clone(set)
-			chosen49[i][peerIndex] = choice49.shouldBroadcastTx(self, peer, txSender)
-		}
-	}
 
-	// Sanity check choices.
+		// Sanity check choices. Here we check that the function selects different peers
-	for i := range chosen49 {
+		// for different transaction senders.
-		c := count(chosen49[i], true)
+		if len(set) != expectedCount {
-		if c == 0 {
+			t.Fatalf("choice49 produced wrong count %d, want %d", len(set), expectedCount)
-			t.Errorf("for tx %d, choice49 chose zero peers", i)
+		}
+		if i > 0 && maps.Equal(set, chosen49[i-1]) {
+			t.Errorf("choice49 for tx %d is equal to tx %d", i, i-1)
 		}
 	}
 
 	// Evaluate choice50 for the same peers and transactions. It should always yield more
 	// peers than choice49, and the chosen set should be a superset of choice49's.
 	for i, txSender := range txsenders {
-		var chosen50 int
+		set := choice50.choosePeers(peers[:50], txSender)
-		for peerIndex, peer := range peers {
+		if len(set) < len(chosen49[i]) {
-			send := choice50.shouldBroadcastTx(self, peer, txSender)
-			if chosen49[i][peerIndex] && !send {
-				t.Errorf("for tx %d, choice50 did not choose peer %d, but choice49 did", i, peerIndex)
-			}
-			if send {
-				chosen50++
-			}
-		}
-		if chosen50 < count(chosen49[i], true) {
 			t.Errorf("for tx %d, choice50 has less peers than choice49", i)
 		}
-	}
+		for p := range chosen49[i] {
-}
+			if _, ok := set[p]; !ok {
-
+				t.Errorf("for tx %d, choice50 did not choose peer %v, but choice49 did", i, p.ID())
-func count[T comparable](s []T, v T) int {
+			}
-	var c int
-	for _, elem := range s {
-		if elem == v {
-			c++
 		}
 	}
-	return c
 }

eth/peerset.go

@@ -19,9 +19,10 @@ package eth
 import (
 	"errors"
 	"fmt"
+	"maps"
+	"slices"
 	"sync"
 
-	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/eth/protocols/eth"
 	"github.com/ethereum/go-ethereum/eth/protocols/snap"
 	"github.com/ethereum/go-ethereum/p2p"
@@ -191,19 +192,12 @@ func (ps *peerSet) peer(id string) *ethPeer {
 	return ps.peers[id]
 }
 
-// peersWithoutTransaction retrieves a list of peers that do not have a given
+// all returns all current peers.
-// transaction in their set of known hashes.
+func (ps *peerSet) all() []*ethPeer {
-func (ps *peerSet) peersWithoutTransaction(hash common.Hash) []*ethPeer {
 	ps.lock.RLock()
 	defer ps.lock.RUnlock()
 
-	list := make([]*ethPeer, 0, len(ps.peers))
+	return slices.Collect(maps.Values(ps.peers))
-	for _, p := range ps.peers {
-		if !p.KnownTransaction(hash) {
-			list = append(list, p)
-		}
-	}
-	return list
 }
 
 // len returns if the current number of `eth` peers in the set. Since the `snap`