Replace periodic push accepted gossip with pull preference gossip for block discovery

chains/manager.go

@@ -205,8 +205,8 @@ type ManagerConfig struct {
 	MeterVMEnabled   bool // Should each VM be wrapped with a MeterVM
 	Metrics          metrics.MultiGatherer
 
-	AcceptedFrontierGossipFrequency time.Duration
-	ConsensusAppConcurrency         int
+	FrontierGossipFrequency time.Duration
+	ConsensusAppConcurrency int
 
 	// Max Time to spend fetching a container and its
 	// ancestors when responding to a GetAncestors
@@ -824,7 +824,7 @@ func (m *manager) createAvalancheChain(
 		ctx,
 		vdrs,
 		msgChan,
-		m.AcceptedFrontierGossipFrequency,
+		m.FrontierGossipFrequency,
 		m.ConsensusAppConcurrency,
 		m.ResourceTracker,
 		validators.UnhandledSubnetConnector, // avalanche chains don't use subnet connector
@@ -1170,7 +1170,7 @@ func (m *manager) createSnowmanChain(
 		ctx,
 		vdrs,
 		msgChan,
-		m.AcceptedFrontierGossipFrequency,
+		m.FrontierGossipFrequency,
 		m.ConsensusAppConcurrency,
 		m.ResourceTracker,
 		subnetConnector,

config/config.go

@@ -60,8 +60,9 @@ const (
 	subnetConfigFileExt  = ".json"
 	ipResolutionTimeout  = 30 * time.Second
 
-	ipcDeprecationMsg      = "IPC API is deprecated"
-	keystoreDeprecationMsg = "keystore API is deprecated"
+	ipcDeprecationMsg                    = "IPC API is deprecated"
+	keystoreDeprecationMsg               = "keystore API is deprecated"
+	acceptedFrontierGossipDeprecationMsg = "push based accepted frontier gossip is deprecated"
 )
 
 var (
@@ -72,6 +73,9 @@ var (
 		IpcsChainIDsKey:       ipcDeprecationMsg,
 		IpcsPathKey:           ipcDeprecationMsg,
 		KeystoreAPIEnabledKey: keystoreDeprecationMsg,
+		ConsensusGossipAcceptedFrontierValidatorSizeKey:    acceptedFrontierGossipDeprecationMsg,
+		ConsensusGossipAcceptedFrontierNonValidatorSizeKey: acceptedFrontierGossipDeprecationMsg,
+		ConsensusGossipAcceptedFrontierPeerSizeKey:         acceptedFrontierGossipDeprecationMsg,
 	}
 
 	errSybilProtectionDisabledStakerWeights   = errors.New("sybil protection disabled weights must be positive")
@@ -1320,9 +1324,9 @@ func GetNodeConfig(v *viper.Viper) (node.Config, error) {
 	}
 
 	// Gossiping
-	nodeConfig.AcceptedFrontierGossipFrequency = v.GetDuration(ConsensusAcceptedFrontierGossipFrequencyKey)
-	if nodeConfig.AcceptedFrontierGossipFrequency < 0 {
-		return node.Config{}, fmt.Errorf("%s must be >= 0", ConsensusAcceptedFrontierGossipFrequencyKey)
+	nodeConfig.FrontierGossipFrequency = v.GetDuration(ConsensusGossipFrontierFrequencyKey)
+	if nodeConfig.FrontierGossipFrequency < 0 {
+		return node.Config{}, fmt.Errorf("%s must be >= 0", ConsensusGossipFrontierFrequencyKey)
 	}
 
 	// App handling

config/flags.go

@@ -177,9 +177,9 @@ func addNodeFlags(fs *pflag.FlagSet) {
 	fs.Duration(BenchlistMinFailingDurationKey, constants.DefaultBenchlistMinFailingDuration, "Minimum amount of time messages to a peer must be failing before the peer is benched")
 
 	// Router
-	fs.Duration(ConsensusAcceptedFrontierGossipFrequencyKey, constants.DefaultAcceptedFrontierGossipFrequency, "Frequency of gossiping accepted frontiers")
 	fs.Uint(ConsensusAppConcurrencyKey, constants.DefaultConsensusAppConcurrency, "Maximum number of goroutines to use when handling App messages on a chain")
 	fs.Duration(ConsensusShutdownTimeoutKey, constants.DefaultConsensusShutdownTimeout, "Timeout before killing an unresponsive chain")
+	fs.Duration(ConsensusGossipFrontierFrequencyKey, constants.DefaultFrontierGossipFrequency, "Frequency of polling and 10%% of gossiping of frontiers")
 	fs.Uint(ConsensusGossipAcceptedFrontierValidatorSizeKey, constants.DefaultConsensusGossipAcceptedFrontierValidatorSize, "Number of validators to gossip to when gossiping accepted frontier")
 	fs.Uint(ConsensusGossipAcceptedFrontierNonValidatorSizeKey, constants.DefaultConsensusGossipAcceptedFrontierNonValidatorSize, "Number of non-validators to gossip to when gossiping accepted frontier")
 	fs.Uint(ConsensusGossipAcceptedFrontierPeerSizeKey, constants.DefaultConsensusGossipAcceptedFrontierPeerSize, "Number of peers to gossip to when gossiping accepted frontier")

config/keys.go

@@ -143,8 +143,9 @@ const (
 	IpcsChainIDsKey                                    = "ipcs-chain-ids"
 	IpcsPathKey                                        = "ipcs-path"
 	MeterVMsEnabledKey                                 = "meter-vms-enabled"
-	ConsensusAcceptedFrontierGossipFrequencyKey        = "consensus-accepted-frontier-gossip-frequency"
 	ConsensusAppConcurrencyKey                         = "consensus-app-concurrency"
+	ConsensusShutdownTimeoutKey                        = "consensus-shutdown-timeout"
+	ConsensusGossipFrontierFrequencyKey                = "consensus-frontier-gossip-frequency"
 	ConsensusGossipAcceptedFrontierValidatorSizeKey    = "consensus-accepted-frontier-gossip-validator-size"
 	ConsensusGossipAcceptedFrontierNonValidatorSizeKey = "consensus-accepted-frontier-gossip-non-validator-size"
 	ConsensusGossipAcceptedFrontierPeerSizeKey         = "consensus-accepted-frontier-gossip-peer-size"
@@ -154,7 +155,6 @@ const (
 	AppGossipValidatorSizeKey                          = "consensus-app-gossip-validator-size"
 	AppGossipNonValidatorSizeKey                       = "consensus-app-gossip-non-validator-size"
 	AppGossipPeerSizeKey                               = "consensus-app-gossip-peer-size"
-	ConsensusShutdownTimeoutKey                        = "consensus-shutdown-timeout"
 	ProposerVMUseCurrentHeightKey                      = "proposervm-use-current-height"
 	FdLimitKey                                         = "fd-limit"
 	IndexEnabledKey                                    = "index-enabled"

node/config.go

@@ -181,8 +181,8 @@ type Config struct {
 	ConsensusRouter          router.Router       `json:"-"`
 	RouterHealthConfig       router.HealthConfig `json:"routerHealthConfig"`
 	ConsensusShutdownTimeout time.Duration       `json:"consensusShutdownTimeout"`
-	// Gossip a container in the accepted frontier every [AcceptedFrontierGossipFrequency]
-	AcceptedFrontierGossipFrequency time.Duration `json:"consensusGossipFreq"`
+	// Gossip and query frontiers every [FrontierGossipFrequency]
+	FrontierGossipFrequency time.Duration `json:"consensusGossipFreq"`
 	// ConsensusAppConcurrency defines the maximum number of goroutines to
 	// handle App messages per chain.
 	ConsensusAppConcurrency int `json:"consensusAppConcurrency"`

node/node.go

@@ -1014,7 +1014,7 @@ func (n *Node) initChainManager(avaxAssetID ids.ID) error {
 		Metrics:                                 n.MetricsGatherer,
 		SubnetConfigs:                           n.Config.SubnetConfigs,
 		ChainConfigs:                            n.Config.ChainConfigs,
-		AcceptedFrontierGossipFrequency:         n.Config.AcceptedFrontierGossipFrequency,
+		FrontierGossipFrequency:                 n.Config.FrontierGossipFrequency,
 		ConsensusAppConcurrency:                 n.Config.ConsensusAppConcurrency,
 		BootstrapMaxTimeGetAncestors:            n.Config.BootstrapMaxTimeGetAncestors,
 		BootstrapAncestorsMaxContainersSent:     n.Config.BootstrapAncestorsMaxContainersSent,

snow/engine/snowman/transitive.go

@@ -31,7 +31,10 @@ import (
 	"github.com/ava-labs/avalanchego/utils/wrappers"
 )
 
-const nonVerifiedCacheSize = 64 * units.MiB
+const (
+	nonVerifiedCacheSize = 64 * units.MiB
+	putGossipPeriod      = 10
+)
 
 var _ Engine = (*Transitive)(nil)
 
@@ -58,7 +61,9 @@ type Transitive struct {
 	common.AppHandler
 	validators.Connector
 
-	RequestID uint32
+	requestID uint32
+
+	gossipCounter int
 
 	// track outstanding preference requests
 	polls poll.Set
@@ -145,6 +150,66 @@ func newTransitive(config Config) (*Transitive, error) {
 	return t, t.metrics.Initialize("", config.Ctx.Registerer)
 }
 
+func (t *Transitive) Gossip(ctx context.Context) error {
+	lastAcceptedID, lastAcceptedHeight := t.Consensus.LastAccepted()
+	if numProcessing := t.Consensus.NumProcessing(); numProcessing == 0 {
+		t.Ctx.Log.Verbo("sampling from validators",
+			zap.Stringer("validators", t.Validators),
+		)
+
+		vdrIDs, err := t.Validators.Sample(t.Ctx.SubnetID, 1)
+		if err != nil {
+			t.Ctx.Log.Error("skipping block gossip",
+				zap.String("reason", "no validators"),
+			)
+			return nil
+		}
+
+		nextHeightToAccept, err := math.Add64(lastAcceptedHeight, 1)
+		if err != nil {
+			t.Ctx.Log.Error("skipping block gossip",
+				zap.String("reason", "block height overflow"),
+				zap.Stringer("blkID", lastAcceptedID),
+				zap.Uint64("lastAcceptedHeight", lastAcceptedHeight),
+				zap.Error(err),
+			)
+			return nil
+		}
+
+		t.requestID++
+		vdrSet := set.Of(vdrIDs...)
+		preferredID := t.Consensus.Preference()
+		t.Sender.SendPullQuery(ctx, vdrSet, t.requestID, preferredID, nextHeightToAccept)
+	} else {
+		t.Ctx.Log.Debug("skipping block gossip",
+			zap.String("reason", "blocks currently processing"),
+			zap.Int("numProcessing", numProcessing),
+		)
+	}
+
+	// TODO: Remove periodic push gossip after v1.11.x is activated
+	t.gossipCounter++
+	t.gossipCounter %= putGossipPeriod
+	if t.gossipCounter > 0 {
+		return nil
+	}
+
+	lastAccepted, err := t.GetBlock(ctx, lastAcceptedID)
+	if err != nil {
+		t.Ctx.Log.Warn("dropping gossip request",
+			zap.String("reason", "block couldn't be loaded"),
+			zap.Stringer("blkID", lastAcceptedID),
+			zap.Error(err),
+		)
+		return nil
+	}
+	t.Ctx.Log.Verbo("gossiping accepted block to the network",
+		zap.Stringer("blkID", lastAcceptedID),
+	)
+	t.Sender.SendGossip(ctx, lastAccepted.Bytes())
+	return nil
+}
+
 func (t *Transitive) Put(ctx context.Context, nodeID ids.NodeID, requestID uint32, blkBytes []byte) error {
 	blk, err := t.VM.ParseBlock(ctx, blkBytes)
 	if err != nil {
@@ -346,28 +411,6 @@ func (*Transitive) Timeout(context.Context) error {
 	return nil
 }
 
-func (t *Transitive) Gossip(ctx context.Context) error {
-	blkID, err := t.VM.LastAccepted(ctx)
-	if err != nil {
-		return err
-	}
-
-	blk, err := t.GetBlock(ctx, blkID)
-	if err != nil {
-		t.Ctx.Log.Warn("dropping gossip request",
-			zap.String("reason", "block couldn't be loaded"),
-			zap.Stringer("blkID", blkID),
-			zap.Error(err),
-		)
-		return nil
-	}
-	t.Ctx.Log.Verbo("gossiping accepted block to the network",
-		zap.Stringer("blkID", blkID),
-	)
-	t.Sender.SendGossip(ctx, blk.Bytes())
-	return nil
-}
-
 func (*Transitive) Halt(context.Context) {}
 
 func (t *Transitive) Shutdown(ctx context.Context) error {
@@ -401,7 +444,7 @@ func (t *Transitive) Context() *snow.ConsensusContext {
 }
 
 func (t *Transitive) Start(ctx context.Context, startReqID uint32) error {
-	t.RequestID = startReqID
+	t.requestID = startReqID
 	lastAcceptedID, err := t.VM.LastAccepted(ctx)
 	if err != nil {
 		return err
@@ -770,14 +813,14 @@ func (t *Transitive) sendRequest(ctx context.Context, nodeID ids.NodeID, blkID i
 		return
 	}
 
-	t.RequestID++
-	t.blkReqs.Add(nodeID, t.RequestID, blkID)
+	t.requestID++
+	t.blkReqs.Add(nodeID, t.requestID, blkID)
 	t.Ctx.Log.Verbo("sending Get request",
 		zap.Stringer("nodeID", nodeID),
-		zap.Uint32("requestID", t.RequestID),
+		zap.Uint32("requestID", t.requestID),
 		zap.Stringer("blkID", blkID),
 	)
-	t.Sender.SendGet(ctx, nodeID, t.RequestID, blkID)
+	t.Sender.SendGet(ctx, nodeID, t.requestID, blkID)
 
 	// Tracks performance statistics
 	t.metrics.numRequests.Set(float64(t.blkReqs.Len()))
@@ -801,6 +844,7 @@ func (t *Transitive) sendQuery(
 		t.Ctx.Log.Error("dropped query for block",
 			zap.String("reason", "insufficient number of validators"),
 			zap.Stringer("blkID", blkID),
+			zap.Int("size", t.Params.K),
 		)
 		return
 	}
@@ -818,21 +862,21 @@ func (t *Transitive) sendQuery(
 	}
 
 	vdrBag := bag.Of(vdrIDs...)
-	t.RequestID++
-	if !t.polls.Add(t.RequestID, vdrBag) {
+	t.requestID++
+	if !t.polls.Add(t.requestID, vdrBag) {
 		t.Ctx.Log.Error("dropped query for block",
 			zap.String("reason", "failed to add poll"),
 			zap.Stringer("blkID", blkID),
-			zap.Uint32("requestID", t.RequestID),
+			zap.Uint32("requestID", t.requestID),
 		)
 		return
 	}
 
 	vdrSet := set.Of(vdrIDs...)
 	if push {
-		t.Sender.SendPushQuery(ctx, vdrSet, t.RequestID, blkBytes, nextHeightToAccept)
+		t.Sender.SendPushQuery(ctx, vdrSet, t.requestID, blkBytes, nextHeightToAccept)
 	} else {
-		t.Sender.SendPullQuery(ctx, vdrSet, t.RequestID, blkID, nextHeightToAccept)
+		t.Sender.SendPullQuery(ctx, vdrSet, t.requestID, blkID, nextHeightToAccept)
 	}
 }
 

snow/engine/snowman/transitive_test.go

@@ -1294,7 +1294,7 @@ func TestEngineUndeclaredDependencyDeadlock(t *testing.T) {
 func TestEngineGossip(t *testing.T) {
 	require := require.New(t)
 
-	_, _, sender, vm, te, gBlk := setupDefaultConfig(t)
+	nodeID, _, sender, vm, te, gBlk := setupDefaultConfig(t)
 
 	vm.LastAcceptedF = func(context.Context) (ids.ID, error) {
 		return gBlk.ID(), nil
@@ -1304,15 +1304,15 @@ func TestEngineGossip(t *testing.T) {
 		return gBlk, nil
 	}
 
-	called := new(bool)
-	sender.SendGossipF = func(_ context.Context, blkBytes []byte) {
-		*called = true
-		require.Equal(gBlk.Bytes(), blkBytes)
+	var calledSendPullQuery bool
+	sender.SendPullQueryF = func(_ context.Context, nodeIDs set.Set[ids.NodeID], _ uint32, _ ids.ID, _ uint64) {
+		calledSendPullQuery = true
+		require.Equal(set.Of(nodeID), nodeIDs)
 	}
 
 	require.NoError(te.Gossip(context.Background()))
 
-	require.True(*called)
+	require.True(calledSendPullQuery)
 }
 
 func TestEngineInvalidBlockIgnoredFromUnexpectedPeer(t *testing.T) {

utils/constants/networking.go

@@ -71,12 +71,12 @@ const (
 	DefaultBenchlistMinFailingDuration = 2*time.Minute + 30*time.Second
 
 	// Router
-	DefaultAcceptedFrontierGossipFrequency                 = 10 * time.Second
 	DefaultConsensusAppConcurrency                         = 2
 	DefaultConsensusShutdownTimeout                        = time.Minute
+	DefaultFrontierGossipFrequency                         = 100 * time.Millisecond
 	DefaultConsensusGossipAcceptedFrontierValidatorSize    = 0
 	DefaultConsensusGossipAcceptedFrontierNonValidatorSize = 0
-	DefaultConsensusGossipAcceptedFrontierPeerSize         = 15
+	DefaultConsensusGossipAcceptedFrontierPeerSize         = 1
 	DefaultConsensusGossipOnAcceptValidatorSize            = 0
 	DefaultConsensusGossipOnAcceptNonValidatorSize         = 0
 	DefaultConsensusGossipOnAcceptPeerSize                 = 10