[network/p2p] Redesign Push Gossip

network/p2p/gossip/gossip.go

@@ -13,9 +13,11 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"go.uber.org/zap"
 
+	"github.com/ava-labs/avalanchego/cache"
 	"github.com/ava-labs/avalanchego/ids"
 	"github.com/ava-labs/avalanchego/network/p2p"
 	"github.com/ava-labs/avalanchego/utils"
+	"github.com/ava-labs/avalanchego/utils/bloom"
 	"github.com/ava-labs/avalanchego/utils/buffer"
 	"github.com/ava-labs/avalanchego/utils/logging"
 )
@@ -24,17 +26,17 @@ const (
 	typeLabel = "type"
 	pushType  = "push"
 	pullType  = "pull"
+
+	defaultGossipableCount = 64
 )
 
 var (
 	_ Gossiper = (*ValidatorGossiper)(nil)
 	_ Gossiper = (*PullGossiper[*testTx])(nil)
 	_ Gossiper = (*NoOpGossiper)(nil)
-	_ Gossiper = (*TestGossiper)(nil)
 
-	_ Accumulator[*testTx] = (*PushGossiper[*testTx])(nil)
-	_ Accumulator[*testTx] = (*NoOpAccumulator[*testTx])(nil)
-	_ Accumulator[*testTx] = (*TestAccumulator[*testTx])(nil)
+	_ Set[*testTx] = (*EmptySet[*testTx])(nil)
+	_ Set[*testTx] = (*FullSet[*testTx])(nil)
 
 	metricLabels = []string{typeLabel}
 	pushLabels   = prometheus.Labels{
@@ -43,6 +45,11 @@ var (
 	pullLabels = prometheus.Labels{
 		typeLabel: pullType,
 	}
+
+	errEmptySetCantAdd          = errors.New("empty set can not add")
+	ErrInvalidDiscardedSize     = errors.New("discarded size cannot be negative")
+	ErrInvalidTargetGossipSize  = errors.New("target gossip size cannot be negative")
+	ErrInvalidRegossipFrequency = errors.New("re-gossip frequency cannot be negative")
 )
 
 // Gossiper gossips Gossipables to other nodes
@@ -51,13 +58,6 @@ type Gossiper interface {
 	Gossip(ctx context.Context) error
 }
 
-// Accumulator allows a caller to accumulate gossipables to be gossiped
-type Accumulator[T Gossipable] interface {
-	Gossiper
-	// Add queues gossipables to be gossiped
-	Add(gossipables ...T)
-}
-
 // ValidatorGossiper only calls [Gossip] if the given node is a validator
 type ValidatorGossiper struct {
 	Gossiper
@@ -73,6 +73,7 @@ type Metrics struct {
 	sentBytes     *prometheus.CounterVec
 	receivedCount *prometheus.CounterVec
 	receivedBytes *prometheus.CounterVec
+	tracking      prometheus.Gauge
 }
 
 // NewMetrics returns a common set of metrics
@@ -101,6 +102,11 @@ func NewMetrics(
 			Name:      "gossip_received_bytes",
 			Help:      "amount of gossip received (bytes)",
 		}, metricLabels),
+		tracking: prometheus.NewGauge(prometheus.GaugeOpts{
+			Namespace: namespace,
+			Name:      "gossip_tracking",
+			Help:      "number of gossipables being tracked",
+		}),
 	}
 	err := utils.Err(
 		metrics.Register(m.sentCount),
@@ -197,17 +203,17 @@ func (p *PullGossiper[_]) handleResponse(
 			continue
 		}
 
-		hash := gossipable.GossipID()
+		gossipID := gossipable.GossipID()
 		p.log.Debug(
 			"received gossip",
 			zap.Stringer("nodeID", nodeID),
-			zap.Stringer("id", hash),
+			zap.Stringer("id", gossipID),
 		)
 		if err := p.set.Add(gossipable); err != nil {
 			p.log.Debug(
 				"failed to add gossip to the known set",
 				zap.Stringer("nodeID", nodeID),
-				zap.Stringer("id", hash),
+				zap.Stringer("id", gossipID),
 				zap.Error(err),
 			)
 			continue
@@ -231,84 +237,190 @@ func (p *PullGossiper[_]) handleResponse(
 }
 
 // NewPushGossiper returns an instance of PushGossiper
-func NewPushGossiper[T Gossipable](marshaller Marshaller[T], client *p2p.Client, metrics Metrics, targetGossipSize int) *PushGossiper[T] {
-	return &PushGossiper[T]{
-		marshaller:       marshaller,
-		client:           client,
-		metrics:          metrics,
-		targetGossipSize: targetGossipSize,
-		pending:          buffer.NewUnboundedDeque[T](0),
+func NewPushGossiper[T Gossipable](
+	marshaller Marshaller[T],
+	mempool Set[T],
+	client *p2p.Client,
+	metrics Metrics,
+	discardedSize int,
+	targetGossipSize int,
+	maxRegossipFrequency time.Duration,
+) (*PushGossiper[T], error) {
+	if discardedSize < 0 {
+		return nil, ErrInvalidDiscardedSize
 	}
+
+	if targetGossipSize < 0 {
+		return nil, ErrInvalidTargetGossipSize
+	}
+
+	if maxRegossipFrequency < 0 {
+		return nil, ErrInvalidRegossipFrequency
+	}
+
+	return &PushGossiper[T]{
+		marshaller:           marshaller,
+		set:                  mempool,
+		client:               client,
+		metrics:              metrics,
+		targetGossipSize:     targetGossipSize,
+		maxRegossipFrequency: maxRegossipFrequency,
+
+		tracking:  make(map[ids.ID]time.Time),
+		pending:   buffer.NewUnboundedDeque[T](0),
+		issued:    buffer.NewUnboundedDeque[T](0),
+		discarded: &cache.LRU[ids.ID, interface{}]{Size: discardedSize},
+	}, nil
 }
 
 // PushGossiper broadcasts gossip to peers randomly in the network
 type PushGossiper[T Gossipable] struct {
-	marshaller       Marshaller[T]
-	client           *p2p.Client
-	metrics          Metrics
-	targetGossipSize int
+	marshaller Marshaller[T]
+	set        Set[T]
+	client     *p2p.Client
+	metrics    Metrics
+
+	targetGossipSize     int
+	maxRegossipFrequency time.Duration
 
-	lock    sync.Mutex
-	pending buffer.Deque[T]
+	lock      sync.Mutex
+	tracking  map[ids.ID]time.Time
+	pending   buffer.Deque[T]
+	issued    buffer.Deque[T]
+	discarded *cache.LRU[ids.ID, interface{}] // discarded attempts to avoid overgossiping transactions that are frequently dropped
 }
 
-// Gossip flushes any queued gossipables
+// Gossip flushes any queued gossipables.
 func (p *PushGossiper[T]) Gossip(ctx context.Context) error {
 	p.lock.Lock()
 	defer p.lock.Unlock()
 
-	if p.pending.Len() == 0 {
+	if len(p.tracking) == 0 {
 		return nil
 	}
 
-	sentBytes := 0
-	gossip := make([][]byte, 0, p.pending.Len())
+	var (
+		sentBytes = 0
+		gossip    = make([][]byte, 0, defaultGossipableCount)
+		now       = time.Now()
+	)
+
+	// Iterate over all pending gossipables (never been sent before).
 	for sentBytes < p.targetGossipSize {
-		gossipable, ok := p.pending.PeekLeft()
+		gossipable, ok := p.pending.PopLeft()
 		if !ok {
 			break
 		}
 
+		// Ensure item is still in the set before we gossip.
+		gossipID := gossipable.GossipID()
+		if !p.set.Has(gossipID) {
+			delete(p.tracking, gossipID)
+			continue
+		}
+
 		bytes, err := p.marshaller.MarshalGossip(gossipable)
 		if err != nil {
-			// remove this item so we don't get stuck in a loop
-			_, _ = p.pending.PopLeft()
+			delete(p.tracking, gossipID)
 			return err
 		}
 
 		gossip = append(gossip, bytes)
 		sentBytes += len(bytes)
-		p.pending.PopLeft()
+		p.issued.PushRight(gossipable)
+		p.tracking[gossipID] = now
 	}
 
+	maxLastGossipTimeToRegossip := now.Add(-p.maxRegossipFrequency)
+
+	// Iterate over all issued gossipables (have been sent before) to fill any
+	// remaining space in gossip batch.
+	for sentBytes < p.targetGossipSize {
+		gossipable, ok := p.issued.PopLeft()
+		if !ok {
+			break
+		}
+
+		// Ensure item is still in the set before we gossip.
+		gossipID := gossipable.GossipID()
+		if !p.set.Has(gossipID) {
+			delete(p.tracking, gossipID)
+			p.discarded.Put(gossipID, nil) // only add to discarded if issued once
+			continue
+		}
+
+		// Ensure we don't attempt to send a gossipable too frequently.
+		lastGossipTime := p.tracking[gossipID]
+		if maxLastGossipTimeToRegossip.Before(lastGossipTime) {
+			// Put the gossipable on the front of the queue to keep items sorted
+			// by last issuance time.
+			p.issued.PushLeft(gossipable)
+			break
+		}
+
+		bytes, err := p.marshaller.MarshalGossip(gossipable)
+		if err != nil {
+			// Should never happen because we've already issued this once.
+			delete(p.tracking, gossipID)
+			return err
+		}
+
+		gossip = append(gossip, bytes)
+		sentBytes += len(bytes)
+		p.issued.PushRight(gossipable)
+		p.tracking[gossipID] = now
+	}
+	p.metrics.tracking.Set(float64(len(p.tracking)))
+
+	// If there is nothing to gossip, we can exit early.
+	if len(gossip) == 0 {
+		return nil
+	}
+
+	// Send gossipables to peers
 	msgBytes, err := MarshalAppGossip(gossip)
 	if err != nil {
 		return err
 	}
-
 	sentCountMetric, err := p.metrics.sentCount.GetMetricWith(pushLabels)
 	if err != nil {
 		return fmt.Errorf("failed to get sent count metric: %w", err)
 	}
-
 	sentBytesMetric, err := p.metrics.sentBytes.GetMetricWith(pushLabels)
 	if err != nil {
 		return fmt.Errorf("failed to get sent bytes metric: %w", err)
 	}
-
 	sentCountMetric.Add(float64(len(gossip)))
 	sentBytesMetric.Add(float64(sentBytes))
-
-	return p.client.AppGossip(ctx, msgBytes)
+	if err := p.client.AppGossip(ctx, msgBytes); err != nil {
+		return fmt.Errorf("failed to gossip: %w", err)
+	}
+	return nil
 }
 
+// Add enqueues new gossipables to be pushed. If a gossiable is already tracked,
+// it is not added again.
 func (p *PushGossiper[T]) Add(gossipables ...T) {
 	p.lock.Lock()
 	defer p.lock.Unlock()
 
+	// Add new gossipables to the pending queue.
+	now := time.Now()
 	for _, gossipable := range gossipables {
-		p.pending.PushRight(gossipable)
+		gossipID := gossipable.GossipID()
+		if _, ok := p.tracking[gossipID]; ok {
+			continue
+		}
+		if _, ok := p.discarded.Get(gossipID); ok {
+			// Pretend that recently discarded transactions were just gossiped.
+			p.tracking[gossipID] = now
+			p.issued.PushRight(gossipable)
+		} else {
+			p.tracking[gossipID] = time.Time{}
+			p.pending.PushRight(gossipable)
+		}
 	}
+	p.metrics.tracking.Set(float64(len(p.tracking)))
 }
 
 // Every calls [Gossip] every [frequency] amount of time.
@@ -335,39 +447,50 @@ func (NoOpGossiper) Gossip(context.Context) error {
 	return nil
 }
 
-type NoOpAccumulator[T Gossipable] struct{}
+type TestGossiper struct {
+	GossipF func(ctx context.Context) error
+}
 
-func (NoOpAccumulator[_]) Gossip(context.Context) error {
+func (t *TestGossiper) Gossip(ctx context.Context) error {
+	return t.GossipF(ctx)
+}
+
+type EmptySet[T Gossipable] struct{}
+
+func (EmptySet[_]) Gossip(context.Context) error {
 	return nil
 }
 
-func (NoOpAccumulator[T]) Add(...T) {}
+func (EmptySet[T]) Add(T) error {
+	return errEmptySetCantAdd
+}
 
-type TestGossiper struct {
-	GossipF func(ctx context.Context) error
+func (EmptySet[T]) Has(ids.ID) bool {
+	return false
 }
 
-func (t *TestGossiper) Gossip(ctx context.Context) error {
-	return t.GossipF(ctx)
+func (EmptySet[T]) Iterate(func(gossipable T) bool) {}
+
+func (EmptySet[_]) GetFilter() ([]byte, []byte) {
+	return bloom.EmptyFilter.Marshal(), ids.Empty[:]
 }
 
-type TestAccumulator[T Gossipable] struct {
-	GossipF func(ctx context.Context) error
-	AddF    func(...T)
+type FullSet[T Gossipable] struct{}
+
+func (FullSet[_]) Gossip(context.Context) error {
+	return nil
 }
 
-func (t TestAccumulator[T]) Gossip(ctx context.Context) error {
-	if t.GossipF == nil {
-		return nil
-	}
+func (FullSet[T]) Add(T) error {
+	return nil
+}
 
-	return t.GossipF(ctx)
+func (FullSet[T]) Has(ids.ID) bool {
+	return true
 }
 
-func (t TestAccumulator[T]) Add(gossipables ...T) {
-	if t.AddF == nil {
-		return
-	}
+func (FullSet[T]) Iterate(func(gossipable T) bool) {}
 
-	t.AddF(gossipables...)
+func (FullSet[_]) GetFilter() ([]byte, []byte) {
+	return bloom.FullFilter.Marshal(), ids.Empty[:]
 }