Integrate stake weighted gossip selection (#511)

* Integrate stake weighted gossip selection

* nit

* merged

* add validity check

go.mod

@@ -4,7 +4,7 @@ go 1.21
 
 require (
 	github.com/VictoriaMetrics/fastcache v1.10.0
-	github.com/ava-labs/avalanchego v1.11.3-codec-cleanup
+	github.com/ava-labs/avalanchego v1.11.3-stake-weighted-gossip.2
 	github.com/cespare/cp v0.1.0
 	github.com/davecgh/go-spew v1.1.1
 	github.com/deckarep/golang-set/v2 v2.1.0

go.sum

@@ -54,8 +54,8 @@ github.com/ajg/form v1.5.1/go.mod h1:uL1WgH+h2mgNtvBq0339dVnzXdBETtL2LeUXaIv25UY
 github.com/allegro/bigcache v1.2.1-0.20190218064605-e24eb225f156 h1:eMwmnE/GDgah4HI848JfFxHt+iPb26b4zyfspmqY0/8=
 github.com/allegro/bigcache v1.2.1-0.20190218064605-e24eb225f156/go.mod h1:Cb/ax3seSYIx7SuZdm2G2xzfwmv3TPSk2ucNfQESPXM=
 github.com/armon/consul-api v0.0.0-20180202201655-eb2c6b5be1b6/go.mod h1:grANhF5doyWs3UAsr3K4I6qtAmlQcZDesFNEHPZAzj8=
-github.com/ava-labs/avalanchego v1.11.3-codec-cleanup h1:vDaeub6mMehtmoYKDNrRf1V2DqGQmbHG2nnJgfvQv0s=
-github.com/ava-labs/avalanchego v1.11.3-codec-cleanup/go.mod h1:oTVnF9idL57J4LM/6RByTmKhI4QvV6OCnF99ysyBljE=
+github.com/ava-labs/avalanchego v1.11.3-stake-weighted-gossip.2 h1:27eTGL+BfWj/QyeWA56rZazRFgO3Ydhq2sOuUjV3geY=
+github.com/ava-labs/avalanchego v1.11.3-stake-weighted-gossip.2/go.mod h1:LZlP3XIzML5IPwxs0jJdMDhTvTkNCakLVsPRHQYm0hI=
 github.com/aymerick/raymond v2.0.3-0.20180322193309-b565731e1464+incompatible/go.mod h1:osfaiScAUVup+UC9Nfq76eWqDhXlp+4UYaA8uhTBO6g=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=

peer/network_test.go

@@ -871,7 +871,7 @@ type testAppSender struct {
 	sendCrossChainAppResponseFn func(ids.ID, uint32, []byte) error
 	sendAppRequestFn            func(context.Context, set.Set[ids.NodeID], uint32, []byte) error
 	sendAppResponseFn           func(ids.NodeID, uint32, []byte) error
-	sendAppGossipFn             func([]byte, int, int, int) error
+	sendAppGossipFn             func(common.SendConfig, []byte) error
 }
 
 func (t testAppSender) SendCrossChainAppRequest(_ context.Context, chainID ids.ID, requestID uint32, appRequestBytes []byte) error {
@@ -882,10 +882,6 @@ func (t testAppSender) SendCrossChainAppResponse(_ context.Context, chainID ids.
 	return t.sendCrossChainAppResponseFn(chainID, requestID, appResponseBytes)
 }
 
-func (t testAppSender) SendAppGossipSpecific(context.Context, set.Set[ids.NodeID], []byte) error {
-	panic("not implemented")
-}
-
 func (t testAppSender) SendAppRequest(ctx context.Context, nodeIDs set.Set[ids.NodeID], requestID uint32, message []byte) error {
 	return t.sendAppRequestFn(ctx, nodeIDs, requestID, message)
 }
@@ -894,8 +890,8 @@ func (t testAppSender) SendAppResponse(_ context.Context, nodeID ids.NodeID, req
 	return t.sendAppResponseFn(nodeID, requestID, message)
 }
 
-func (t testAppSender) SendAppGossip(_ context.Context, message []byte, numValidators int, numNonValidators int, numPeers int) error {
-	return t.sendAppGossipFn(message, numValidators, numNonValidators, numPeers)
+func (t testAppSender) SendAppGossip(_ context.Context, config common.SendConfig, message []byte) error {
+	return t.sendAppGossipFn(config, message)
 }
 
 func (t testAppSender) SendAppError(ctx context.Context, nodeID ids.NodeID, requestID uint32, errorCode int32, errorMessage string) error {

plugin/evm/config.go

@@ -35,6 +35,7 @@ const (
 	defaultMaxBlocksPerRequest                        = 0 // Default to no maximum on the number of blocks per getLogs request
 	defaultContinuousProfilerFrequency                = 15 * time.Minute
 	defaultContinuousProfilerMaxFiles                 = 5
+	defaultPushGossipPercentStake                     = .9
 	defaultPushGossipNumValidators                    = 100
 	defaultPushGossipNumPeers                         = 0
 	defaultPushRegossipNumValidators                  = 10
@@ -155,6 +156,7 @@ type Config struct {
 	KeystoreInsecureUnlockAllowed bool   `json:"keystore-insecure-unlock-allowed"`
 
 	// Gossip Settings
+	PushGossipPercentStake    float64  `json:"push-gossip-percent-stake"`
 	PushGossipNumValidators   int      `json:"push-gossip-num-validators"`
 	PushGossipNumPeers        int      `json:"push-gossip-num-peers"`
 	PushRegossipNumValidators int      `json:"push-regossip-num-validators"`
@@ -259,6 +261,7 @@ func (c *Config) SetDefaults() {
 	c.CommitInterval = defaultCommitInterval
 	c.SnapshotWait = defaultSnapshotWait
 	c.RegossipFrequency.Duration = defaultTxRegossipFrequency
+	c.PushGossipPercentStake = defaultPushGossipPercentStake
 	c.PushGossipNumValidators = defaultPushGossipNumValidators
 	c.PushGossipNumPeers = defaultPushGossipNumPeers
 	c.PushRegossipNumValidators = defaultPushRegossipNumValidators
@@ -315,6 +318,9 @@ func (c *Config) Validate() error {
 		return fmt.Errorf("cannot use commit interval of 0 with pruning enabled")
 	}
 
+	if c.PushGossipPercentStake < 0 || c.PushGossipPercentStake > 1 {
+		return fmt.Errorf("push-gossip-percent-stake is %f but must be in the range [0, 1]", c.PushGossipPercentStake)
+	}
 	return nil
 }
 

plugin/evm/gossiper_atomic_gossiping_test.go

@@ -14,6 +14,8 @@ import (
 	"github.com/ava-labs/avalanchego/utils/set"
 	"github.com/stretchr/testify/assert"
 
+	commonEng "github.com/ava-labs/avalanchego/snow/engine/common"
+
 	"github.com/ava-labs/coreth/plugin/evm/message"
 )
 
@@ -35,7 +37,7 @@ func TestMempoolAtmTxsAppGossipHandling(t *testing.T) {
 		txRequested    bool
 	)
 	sender.CantSendAppGossip = false
-	sender.SendAppGossipF = func(context.Context, []byte, int, int, int) error {
+	sender.SendAppGossipF = func(context.Context, commonEng.SendConfig, []byte) error {
 		txGossipedLock.Lock()
 		defer txGossipedLock.Unlock()
 
@@ -122,7 +124,7 @@ func TestMempoolAtmTxsAppGossipHandlingDiscardedTx(t *testing.T) {
 		txRequested    bool
 	)
 	sender.CantSendAppGossip = false
-	sender.SendAppGossipF = func(context.Context, []byte, int, int, int) error {
+	sender.SendAppGossipF = func(context.Context, commonEng.SendConfig, []byte) error {
 		txGossipedLock.Lock()
 		defer txGossipedLock.Unlock()
 

plugin/evm/gossiper_eth_gossiping_test.go

@@ -17,6 +17,8 @@ import (
 	"github.com/ava-labs/avalanchego/ids"
 	"github.com/ava-labs/avalanchego/utils/set"
 
+	commonEng "github.com/ava-labs/avalanchego/snow/engine/common"
+
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/rlp"
@@ -105,7 +107,7 @@ func TestMempoolEthTxsAppGossipHandling(t *testing.T) {
 		return nil
 	}
 	wg.Add(1)
-	sender.SendAppGossipF = func(context.Context, []byte, int, int, int) error {
+	sender.SendAppGossipF = func(context.Context, commonEng.SendConfig, []byte) error {
 		wg.Done()
 		return nil
 	}

plugin/evm/syncervm_test.go

@@ -116,7 +116,7 @@ func TestStateSyncToggleEnabledToDisabled(t *testing.T) {
 
 	syncDisabledVM := &VM{}
 	appSender := &commonEng.SenderTest{T: t}
-	appSender.SendAppGossipF = func(context.Context, []byte, int, int, int) error { return nil }
+	appSender.SendAppGossipF = func(context.Context, commonEng.SendConfig, []byte) error { return nil }
 	appSender.SendAppRequestF = func(ctx context.Context, nodeSet set.Set[ids.NodeID], requestID uint32, request []byte) error {
 		nodeID, hasItem := nodeSet.Pop()
 		if !hasItem {

plugin/evm/tx_gossip_test.go

@@ -93,7 +93,12 @@ func TestEthTxGossip(t *testing.T) {
 		return 0, nil
 	}
 	validatorState.GetValidatorSetF = func(context.Context, uint64, ids.ID) (map[ids.NodeID]*validators.GetValidatorOutput, error) {
-		return map[ids.NodeID]*validators.GetValidatorOutput{requestingNodeID: nil}, nil
+		return map[ids.NodeID]*validators.GetValidatorOutput{
+			requestingNodeID: {
+				NodeID: requestingNodeID,
+				Weight: 1,
+			},
+		}, nil
 	}
 
 	// Ask the VM for any new transactions. We should get nothing at first.
@@ -220,7 +225,12 @@ func TestAtomicTxGossip(t *testing.T) {
 		return 0, nil
 	}
 	validatorState.GetValidatorSetF = func(context.Context, uint64, ids.ID) (map[ids.NodeID]*validators.GetValidatorOutput, error) {
-		return map[ids.NodeID]*validators.GetValidatorOutput{requestingNodeID: nil}, nil
+		return map[ids.NodeID]*validators.GetValidatorOutput{
+			requestingNodeID: {
+				NodeID: requestingNodeID,
+				Weight: 1,
+			},
+		}, nil
 	}
 
 	// Ask the VM for any new transactions. We should get nothing at first.
@@ -296,6 +306,14 @@ func TestEthTxPushGossipOutbound(t *testing.T) {
 	require := require.New(t)
 	ctx := context.Background()
 	snowCtx := utils.TestSnowContext()
+	snowCtx.ValidatorState = &validators.TestState{
+		GetCurrentHeightF: func(context.Context) (uint64, error) {
+			return 0, nil
+		},
+		GetValidatorSetF: func(context.Context, uint64, ids.ID) (map[ids.NodeID]*validators.GetValidatorOutput, error) {
+			return nil, nil
+		},
+	}
 	sender := &common.FakeSender{
 		SentAppGossip: make(chan []byte, 1),
 	}

plugin/evm/vm.go

@@ -1099,8 +1099,9 @@ func (vm *VM) initBlockBuilding() error {
 	}
 
 	pushGossipParams := gossip.BranchingFactor{
-		Validators: vm.config.PushGossipNumValidators,
-		Peers:      vm.config.PushGossipNumPeers,
+		StakePercentage: vm.config.PushGossipPercentStake,
+		Validators:      vm.config.PushGossipNumValidators,
+		Peers:           vm.config.PushGossipNumPeers,
 	}
 	pushRegossipParams := gossip.BranchingFactor{
 		Validators: vm.config.PushRegossipNumValidators,
@@ -1112,6 +1113,7 @@ func (vm *VM) initBlockBuilding() error {
 		ethTxPushGossiper, err = gossip.NewPushGossiper[*GossipEthTx](
 			ethTxGossipMarshaller,
 			ethTxPool,
+			vm.validators,
 			ethTxGossipClient,
 			ethTxGossipMetrics,
 			pushGossipParams,
@@ -1130,6 +1132,7 @@ func (vm *VM) initBlockBuilding() error {
 		vm.atomicTxPushGossiper, err = gossip.NewPushGossiper[*GossipAtomicTx](
 			atomicTxGossipMarshaller,
 			vm.mempool,
+			vm.validators,
 			atomicTxGossipClient,
 			atomicTxGossipMetrics,
 			pushGossipParams,

plugin/evm/vm_test.go

@@ -253,7 +253,7 @@ func GenesisVM(t *testing.T,
 	ctx, dbManager, genesisBytes, issuer, m := setupGenesis(t, genesisJSON)
 	appSender := &commonEng.SenderTest{T: t}
 	appSender.CantSendAppGossip = true
-	appSender.SendAppGossipF = func(context.Context, []byte, int, int, int) error { return nil }
+	appSender.SendAppGossipF = func(context.Context, commonEng.SendConfig, []byte) error { return nil }
 	err := vm.Initialize(
 		context.Background(),
 		ctx,
@@ -3409,7 +3409,7 @@ func TestConfigureLogLevel(t *testing.T) {
 			ctx, dbManager, genesisBytes, issuer, _ := setupGenesis(t, test.genesisJSON)
 			appSender := &commonEng.SenderTest{T: t}
 			appSender.CantSendAppGossip = true
-			appSender.SendAppGossipF = func(context.Context, []byte, int, int, int) error { return nil }
+			appSender.SendAppGossipF = func(context.Context, commonEng.SendConfig, []byte) error { return nil }
 			err := vm.Initialize(
 				context.Background(),
 				ctx,

scripts/versions.sh

@@ -6,4 +6,4 @@
 set -euo pipefail
 
 # Don't export them as they're used in the context of other calls
-avalanche_version=${AVALANCHE_VERSION:-'v1.11.3-codec-cleanup'}
+avalanche_version=${AVALANCHE_VERSION:-'v1.11.3-stake-weighted-gossip.2'}