Allow configuring push gossip to send txs to validators by stake

go.mod

@@ -10,7 +10,7 @@ go 1.21
 require (
 	github.com/DataDog/zstd v1.5.2
 	github.com/NYTimes/gziphandler v1.1.1
-	github.com/ava-labs/coreth v0.13.2-rc.0
+	github.com/ava-labs/coreth v0.13.2-stake-sampling.2
 	github.com/ava-labs/ledger-avalanche/go v0.0.0-20231102202641-ae2ebdaeac34
 	github.com/btcsuite/btcd/btcutil v1.1.3
 	github.com/cockroachdb/pebble v0.0.0-20230209160836-829675f94811
@@ -68,7 +68,6 @@ require (
 )
 
 require (
-	github.com/BurntSushi/toml v1.2.1 // indirect
 	github.com/FactomProject/basen v0.0.0-20150613233007-fe3947df716e // indirect
 	github.com/FactomProject/btcutilecc v0.0.0-20130527213604-d3a63a5752ec // indirect
 	github.com/VictoriaMetrics/fastcache v1.10.0 // indirect

go.sum

@@ -38,8 +38,8 @@ cloud.google.com/go/storage v1.14.0/go.mod h1:GrKmX003DSIwi9o29oFT7YDnHYwZoctc3f
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
 github.com/AndreasBriese/bbloom v0.0.0-20190306092124-e2d15f34fcf9/go.mod h1:bOvUY6CB00SOBii9/FifXqc0awNKxLFCL/+pkDPuyl8=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
-github.com/BurntSushi/toml v1.2.1 h1:9F2/+DoOYIOksmaJFPw1tGFy1eDnIJXg+UHjuD8lTak=
-github.com/BurntSushi/toml v1.2.1/go.mod h1:CxXYINrC8qIiEnFrOxCa7Jy5BFHlXnUU2pbicEuybxQ=
+github.com/BurntSushi/toml v1.3.2 h1:o7IhLm0Msx3BaB+n3Ag7L8EVlByGnpq14C4YWiu/gL8=
+github.com/BurntSushi/toml v1.3.2/go.mod h1:CxXYINrC8qIiEnFrOxCa7Jy5BFHlXnUU2pbicEuybxQ=
 github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo=
 github.com/CloudyKit/fastprinter v0.0.0-20200109182630-33d98a066a53/go.mod h1:+3IMCy2vIlbG1XG/0ggNQv0SvxCAIpPM5b1nCz56Xno=
 github.com/CloudyKit/jet/v3 v3.0.0/go.mod h1:HKQPgSJmdK8hdoAbKUUWajkHyHo4RaU5rMdUywE7VMo=
@@ -63,8 +63,8 @@ github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRF
 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/coreth v0.13.2-rc.0 h1:D1BqbxAMuMmagueDYOzET8PS1qZxbqnvRhP5eIZ1On8=
-github.com/ava-labs/coreth v0.13.2-rc.0/go.mod h1:Mpdw41yvGdb8IJOIpcPZYz5O3wyprVwHPV02J8JvdeA=
+github.com/ava-labs/coreth v0.13.2-stake-sampling.2 h1:tFnjwSEIMeh+dkWcB52c//wTVngZiJ2Rf6TW4ZcsCCc=
+github.com/ava-labs/coreth v0.13.2-stake-sampling.2/go.mod h1:3AtWfbwSDORoaWEui9o5X2WmPlVbksm3zWuUzN9q/60=
 github.com/ava-labs/ledger-avalanche/go v0.0.0-20231102202641-ae2ebdaeac34 h1:mg9Uw6oZFJKytJxgxnl3uxZOs/SB8CVHg6Io4Tf99Zc=
 github.com/ava-labs/ledger-avalanche/go v0.0.0-20231102202641-ae2ebdaeac34/go.mod h1:pJxaT9bUgeRNVmNRgtCHb7sFDIRKy7CzTQVi8gGNT6g=
 github.com/aymerick/raymond v2.0.3-0.20180322193309-b565731e1464+incompatible/go.mod h1:osfaiScAUVup+UC9Nfq76eWqDhXlp+4UYaA8uhTBO6g=

network/p2p/gossip/gossip.go

@@ -21,6 +21,7 @@ import (
 	"github.com/ava-labs/avalanchego/utils/bloom"
 	"github.com/ava-labs/avalanchego/utils/buffer"
 	"github.com/ava-labs/avalanchego/utils/logging"
+	"github.com/ava-labs/avalanchego/utils/set"
 )
 
 const (
@@ -89,6 +90,7 @@ type Metrics struct {
 	receivedBytes           *prometheus.CounterVec
 	tracking                *prometheus.GaugeVec
 	trackingLifetimeAverage prometheus.Gauge
+	topValidators           *prometheus.GaugeVec
 }
 
 // NewMetrics returns a common set of metrics
@@ -127,6 +129,11 @@ func NewMetrics(
 			Name:      "gossip_tracking_lifetime_average",
 			Help:      "average duration a gossipable has been tracked (ns)",
 		}),
+		topValidators: prometheus.NewGaugeVec(prometheus.GaugeOpts{
+			Namespace: namespace,
+			Name:      "top_validators",
+			Help:      "number of validators gossipables are sent to due to stake",
+		}, metricLabels),
 	}
 	err := utils.Err(
 		metrics.Register(m.sentCount),
@@ -135,6 +142,7 @@ func NewMetrics(
 		metrics.Register(m.receivedBytes),
 		metrics.Register(m.tracking),
 		metrics.Register(m.trackingLifetimeAverage),
+		metrics.Register(m.topValidators),
 	)
 	return m, err
 }
@@ -262,6 +270,7 @@ func (p *PullGossiper[_]) handleResponse(
 func NewPushGossiper[T Gossipable](
 	marshaller Marshaller[T],
 	mempool Set[T],
+	validators p2p.ValidatorSubset,
 	client *p2p.Client,
 	metrics Metrics,
 	gossipParams BranchingFactor,
@@ -288,6 +297,7 @@ func NewPushGossiper[T Gossipable](
 	return &PushGossiper[T]{
 		marshaller:           marshaller,
 		set:                  mempool,
+		validators:           validators,
 		client:               client,
 		metrics:              metrics,
 		gossipParams:         gossipParams,
@@ -306,6 +316,7 @@ func NewPushGossiper[T Gossipable](
 type PushGossiper[T Gossipable] struct {
 	marshaller Marshaller[T]
 	set        Set[T]
+	validators p2p.ValidatorSubset
 	client     *p2p.Client
 	metrics    Metrics
 
@@ -323,9 +334,20 @@ type PushGossiper[T Gossipable] struct {
 }
 
 type BranchingFactor struct {
-	Validators    int
+	// StakePercentage determines the percentage of stake that should have
+	// gossip sent to based on the inverse CDF of stake weights. This value does
+	// not account for the connectivity of the nodes.
+	StakePercentage float64
+	// Validators specifies the number of connected validators, in addition to
+	// any validators sent from the StakePercentage parameter, to send gossip
+	// to. These validators are sampled uniformly rather than by stake.
+	Validators int
+	// NonValidators specifies the number of connected non-validators to send
+	// gossip to.
 	NonValidators int
-	Peers         int
+	// Peers specifies the number of connected validators or non-validators, in
+	// addition to the number sent due to other configs, to send gossip to.
+	Peers int
 }
 
 func (b *BranchingFactor) Verify() error {
@@ -372,6 +394,7 @@ func (p *PushGossiper[T]) Gossip(ctx context.Context) error {
 		p.toGossip,
 		p.toRegossip,
 		&cache.Empty[ids.ID, struct{}]{}, // Don't mark dropped unsent transactions as discarded
+		unsentLabels,
 	); err != nil {
 		return fmt.Errorf("unexpected error during gossip: %w", err)
 	}
@@ -383,6 +406,7 @@ func (p *PushGossiper[T]) Gossip(ctx context.Context) error {
 		p.toRegossip,
 		p.toRegossip,
 		p.discarded, // Mark dropped sent transactions as discarded
+		sentLabels,
 	); err != nil {
 		return fmt.Errorf("unexpected error during regossip: %w", err)
 	}
@@ -396,6 +420,7 @@ func (p *PushGossiper[T]) gossip(
 	toGossip buffer.Deque[T],
 	toRegossip buffer.Deque[T],
 	discarded cache.Cacher[ids.ID, struct{}],
+	metricsLabels prometheus.Labels,
 ) error {
 	var (
 		sentBytes                   = 0
@@ -450,6 +475,9 @@ func (p *PushGossiper[T]) gossip(
 	if err != nil {
 		return err
 	}
+
+	validatorsByStake := p.validators.Top(ctx, gossipParams.StakePercentage)
+
 	sentCountMetric, err := p.metrics.sentCount.GetMetricWith(pushLabels)
 	if err != nil {
 		return fmt.Errorf("failed to get sent count metric: %w", err)
@@ -458,12 +486,18 @@ func (p *PushGossiper[T]) gossip(
 	if err != nil {
 		return fmt.Errorf("failed to get sent bytes metric: %w", err)
 	}
+	topValidatorsMetric, err := p.metrics.topValidators.GetMetricWith(metricsLabels)
+	if err != nil {
+		return fmt.Errorf("failed to get top validators metric: %w", err)
+	}
 	sentCountMetric.Add(float64(len(gossip)))
 	sentBytesMetric.Add(float64(sentBytes))
+	topValidatorsMetric.Set(float64(len(validatorsByStake)))
 
 	return p.client.AppGossip(
 		ctx,
 		common.SendConfig{
+			NodeIDs:       set.Of(validatorsByStake...),
 			Validators:    gossipParams.Validators,
 			NonValidators: gossipParams.NonValidators,
 			Peers:         gossipParams.Peers,

network/p2p/gossip/gossip_test.go

@@ -18,6 +18,8 @@ import (
 	"github.com/ava-labs/avalanchego/network/p2p"
 	"github.com/ava-labs/avalanchego/proto/pb/sdk"
 	"github.com/ava-labs/avalanchego/snow/engine/common"
+	"github.com/ava-labs/avalanchego/snow/validators"
+	"github.com/ava-labs/avalanchego/utils/constants"
 	"github.com/ava-labs/avalanchego/utils/logging"
 	"github.com/ava-labs/avalanchego/utils/set"
 	"github.com/ava-labs/avalanchego/utils/units"
@@ -359,6 +361,7 @@ func TestPushGossiperNew(t *testing.T) {
 				nil,
 				nil,
 				nil,
+				nil,
 				Metrics{},
 				tt.gossipParams,
 				tt.regossipParams,
@@ -517,6 +520,20 @@ func TestPushGossiper(t *testing.T) {
 			)
 			require.NoError(err)
 			client := network.NewClient(0)
+			validators := p2p.NewValidators(
+				&p2p.Peers{},
+				logging.NoLog{},
+				constants.PrimaryNetworkID,
+				&validators.TestState{
+					GetCurrentHeightF: func(context.Context) (uint64, error) {
+						return 1, nil
+					},
+					GetValidatorSetF: func(context.Context, uint64, ids.ID) (map[ids.NodeID]*validators.GetValidatorOutput, error) {
+						return nil, nil
+					},
+				},
+				time.Hour,
+			)
 			metrics, err := NewMetrics(prometheus.NewRegistry(), "")
 			require.NoError(err)
 			marshaller := testMarshaller{}
@@ -529,6 +546,7 @@ func TestPushGossiper(t *testing.T) {
 			gossiper, err := NewPushGossiper[*testTx](
 				marshaller,
 				FullSet[*testTx]{},
+				validators,
 				client,
 				metrics,
 				BranchingFactor{

network/p2p/network_test.go

@@ -561,7 +561,10 @@ func TestNodeSamplerClientOption(t *testing.T) {
 					},
 					GetValidatorSetF: func(context.Context, uint64, ids.ID) (map[ids.NodeID]*validators.GetValidatorOutput, error) {
 						return map[ids.NodeID]*validators.GetValidatorOutput{
-							nodeID1: nil,
+							nodeID1: {
+								NodeID: nodeID1,
+								Weight: 1,
+							},
 						}, nil
 					},
 				}
@@ -581,7 +584,10 @@ func TestNodeSamplerClientOption(t *testing.T) {
 					},
 					GetValidatorSetF: func(context.Context, uint64, ids.ID) (map[ids.NodeID]*validators.GetValidatorOutput, error) {
 						return map[ids.NodeID]*validators.GetValidatorOutput{
-							nodeID1: nil,
+							nodeID1: {
+								NodeID: nodeID1,
+								Weight: 1,
+							},
 						}, nil
 					},
 				}