feat: gossip improvements

pkg/hive/gossip_buffer.go

@@ -0,0 +1,101 @@
+// Copyright 2026 The Swarm Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package hive
+
+import (
+	"maps"
+	"slices"
+	"sync"
+	"time"
+
+	"github.com/ethersphere/bee/v2/pkg/swarm"
+)
+
+const (
+	defaultGossipCoalesceInterval = time.Second
+	// coalesceThreshold: gossips with fewer peers are buffered; larger
+	// (already-batched) messages are dispatched immediately.
+	coalesceThreshold = 2
+)
+
+// gossipBuffer accumulates single-peer outbound gossip per addressee so it can be
+// flushed as one batched message.
+type gossipBuffer struct {
+	mu       sync.Mutex
+	pending  map[string]map[string]swarm.Address // addressee key -> peer key -> peer
+	interval time.Duration
+	maxBatch int
+}
+
+type gossipBatch struct {
+	addressee swarm.Address
+	peers     []swarm.Address
+}
+
+func newGossipBuffer(interval time.Duration, maxBatch int) *gossipBuffer {
+	if interval == 0 {
+		interval = defaultGossipCoalesceInterval
+	}
+	return &gossipBuffer{
+		pending:  make(map[string]map[string]swarm.Address),
+		interval: interval,
+		maxBatch: maxBatch,
+	}
+}
+
+// stagePeers buffers peers for the addressee. If the buffer reaches maxBatch it is
+// removed and returned so the caller can flush it immediately.
+func (b *gossipBuffer) stagePeers(addressee swarm.Address, peers ...swarm.Address) (flushPeers []swarm.Address, flush bool) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	key := addressee.ByteString()
+	peerSet, ok := b.pending[key]
+	if !ok {
+		peerSet = make(map[string]swarm.Address)
+		b.pending[key] = peerSet
+	}
+	for _, p := range peers {
+		peerSet[p.ByteString()] = p
+	}
+
+	if len(peerSet) >= b.maxBatch {
+		delete(b.pending, key)
+		return slices.Collect(maps.Values(peerSet)), true
+	}
+	return nil, false
+}
+
+// takeAll removes and returns all buffered entries.
+func (b *gossipBuffer) takeAll() []gossipBatch {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	if len(b.pending) == 0 {
+		return nil
+	}
+
+	out := make([]gossipBatch, 0, len(b.pending))
+	for key, peerSet := range b.pending {
+		out = append(out, gossipBatch{
+			addressee: swarm.NewAddress([]byte(key)),
+			peers:     slices.Collect(maps.Values(peerSet)),
+		})
+	}
+	b.pending = make(map[string]map[string]swarm.Address)
+	return out
+}
+
+func (b *gossipBuffer) clearAddressee(addressee swarm.Address) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	delete(b.pending, addressee.ByteString())
+}
+
+func (b *gossipBuffer) pendingAddressees() int {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return len(b.pending)
+}

pkg/hive/gossip_buffer_test.go

@@ -0,0 +1,67 @@
+// Copyright 2026 The Swarm Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package hive
+
+import (
+	"testing"
+	"time"
+
+	"github.com/ethersphere/bee/v2/pkg/swarm"
+)
+
+func TestGossipBufferAddAndTakeAll(t *testing.T) {
+	t.Parallel()
+
+	b := newGossipBuffer(time.Second, maxBatchSize)
+	addressee := swarm.RandAddress(t)
+	peer1 := swarm.RandAddress(t)
+	peer2 := swarm.RandAddress(t)
+
+	if pending := b.takeAll(); len(pending) != 0 {
+		t.Fatalf("want no pending entries, got %d", len(pending))
+	}
+
+	if _, flush := b.stagePeers(addressee, peer1); flush {
+		t.Fatal("unexpected immediate flush")
+	}
+
+	if _, flush := b.stagePeers(addressee, peer2); flush {
+		t.Fatal("unexpected immediate flush")
+	}
+
+	pending := b.takeAll()
+	if len(pending) != 1 {
+		t.Fatalf("want 1 pending entry, got %d", len(pending))
+	}
+	if got := len(pending[0].peers); got != 2 {
+		t.Fatalf("want 2 coalesced peers, got %d", got)
+	}
+	if !pending[0].addressee.Equal(addressee) {
+		t.Fatal("unexpected addressee in pending batch")
+	}
+
+	if pending := b.takeAll(); len(pending) != 0 {
+		t.Fatalf("want empty buffer after takeAll, got %d pending", len(pending))
+	}
+}
+
+func TestGossipBufferMaxBatchFlush(t *testing.T) {
+	t.Parallel()
+
+	b := newGossipBuffer(time.Second, 2)
+	addressee := swarm.RandAddress(t)
+
+	b.stagePeers(addressee, swarm.RandAddress(t))
+	flushPeers, flush := b.stagePeers(addressee, swarm.RandAddress(t))
+	if !flush {
+		t.Fatal("want immediate flush at maxBatch")
+	}
+	if got := len(flushPeers); got != 2 {
+		t.Fatalf("want 2 peers in full batch, got %d", got)
+	}
+	if pending := b.takeAll(); len(pending) != 0 {
+		t.Fatalf("want empty buffer after maxBatch flush, got %d pending", len(pending))
+	}
+}

pkg/hive/hive.go

@@ -57,6 +57,11 @@ var (
 	ErrRateLimitExceeded = errors.New("rate limit exceeded")
 )
 
+const (
+	coalesceFlushReasonTimer    = "timer"
+	coalesceFlushReasonMaxBatch = "max_batch"
+)
+
 // Options configures hive.Service at construction. Chequebook fields are
 // optional: a nil ChequebookVerifier disables the verification gate (and
 // records without a chequebook are accepted); a nil ChequebookStorer means
@@ -66,6 +71,8 @@ type Options struct {
 	AllowPrivateCIDRs  bool
 	ChequebookVerifier chequebook.Verifier
 	ChequebookStorer   ChequebookStorer
+
+	GossipCoalesceInterval time.Duration
 }
 
 type Service struct {
@@ -90,6 +97,7 @@ type Service struct {
 	// chequebook are dropped.
 	chequebookVerifier chequebook.Verifier
 	chequebookStorer   ChequebookStorer
+	gossipBuf          *gossipBuffer
 }
 
 func New(streamer p2p.Streamer, addressbook addressbook.GetPutter, networkID uint64, overlay swarm.Address, logger log.Logger, o Options) *Service {
@@ -112,9 +120,12 @@ func New(streamer p2p.Streamer, addressbook addressbook.GetPutter, networkID uin
 		chequebookStorer:   o.ChequebookStorer,
 	}
 
+	svc.gossipBuf = newGossipBuffer(o.GossipCoalesceInterval, maxBatchSize)
+
 	if !o.BootnodeMode {
 		svc.startCheckPeersHandler()
 	}
+	svc.startGossipCoalescer()
 
 	return svc
 }
@@ -136,34 +147,77 @@ func (s *Service) Protocol() p2p.ProtocolSpec {
 
 var ErrShutdownInProgress = errors.New("shutdown in progress")
 
+// BroadcastPeers sends peer gossip to the addressee. Calls with fewer than
+// coalesceThreshold peers are buffered and flushed asynchronously; errors
+// during deferred dispatch are logged but not returned to the caller.
+// Calls with coalesceThreshold or more peers are sent immediately.
 func (s *Service) BroadcastPeers(ctx context.Context, addressee swarm.Address, peers ...swarm.Address) error {
-	maxSize := maxBatchSize
+	if len(peers) == 0 {
+		return nil
+	}
+
 	s.metrics.BroadcastPeers.Inc()
 	s.metrics.BroadcastPeersPeers.Add(float64(len(peers)))
 
+	// Already-batched messages go out immediately; single-peer gossips are coalesced.
+	if len(peers) >= coalesceThreshold {
+		s.metrics.GossipCoalesceImmediatePeers.Add(float64(len(peers)))
+		s.logger.Debug("gossip immediate send", "addressee", addressee, "peer_count", len(peers))
+		return s.broadcastNow(ctx, addressee, false, peers...)
+	}
+
+	select {
+	case <-s.quit:
+		return ErrShutdownInProgress
+	default:
+	}
+
+	s.metrics.GossipCoalesceBufferedPeers.Add(float64(len(peers)))
+	s.logger.Debug("gossip buffered", "addressee", addressee, "peer_count", len(peers))
+
+	// Buffer; if it just filled up, flush it synchronously while still in the call
+	if flushPeers, flush := s.gossipBuf.stagePeers(addressee, peers...); flush {
+		s.recordCoalesceFlush(coalesceFlushReasonMaxBatch, addressee, flushPeers)
+		s.setCoalesceBufferGauge()
+		return s.broadcastNow(ctx, addressee, true, flushPeers...)
+	}
+	s.setCoalesceBufferGauge()
+	return nil
+}
+
+// broadcastNow performs the synchronous, rate-limited, batched send.
+func (s *Service) broadcastNow(ctx context.Context, addressee swarm.Address, coalesced bool, peers ...swarm.Address) error {
+	maxSize := maxBatchSize
+
 	for len(peers) > 0 {
 		if maxSize > len(peers) {
 			maxSize = len(peers)
 		}
 
-		// If broadcasting limit is exceeded, return early
 		if !s.outLimiter.Allow(addressee.ByteString(), maxSize) {
+			if coalesced {
+				s.metrics.GossipCoalesceDropped.Add(float64(len(peers)))
+			}
 			return nil
 		}
 
 		select {
+		case <-ctx.Done():
+			return ctx.Err()
 		case <-s.quit:
 			return ErrShutdownInProgress
 		default:
 		}
 
 		if err := s.sendPeers(ctx, addressee, peers[:maxSize]); err != nil {
+			if coalesced {
+				s.metrics.GossipCoalesceDropped.Add(float64(len(peers)))
+			}
 			return err
 		}
 
 		peers = peers[maxSize:]
 	}
-
 	return nil
 }
 
@@ -296,9 +350,55 @@ func (s *Service) peersHandler(ctx context.Context, peer p2p.Peer, stream p2p.St
 func (s *Service) disconnect(peer p2p.Peer) error {
 	s.inLimiter.Clear(peer.Address.ByteString())
 	s.outLimiter.Clear(peer.Address.ByteString())
+	s.gossipBuf.clearAddressee(peer.Address)
+	s.setCoalesceBufferGauge()
 	return nil
 }
 
+func (s *Service) startGossipCoalescer() {
+	s.wg.Go(func() {
+		ticker := time.NewTicker(s.gossipBuf.interval)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ticker.C:
+				for _, batch := range s.gossipBuf.takeAll() {
+					s.flushGossipBatch(batch.addressee, batch.peers, coalesceFlushReasonTimer)
+				}
+				s.setCoalesceBufferGauge()
+			case <-s.quit:
+				return
+			}
+		}
+	})
+}
+
+func (s *Service) flushGossipBatch(addressee swarm.Address, peers []swarm.Address, reason string) {
+	s.recordCoalesceFlush(reason, addressee, peers)
+
+	ctx, cancel := context.WithTimeout(context.Background(), messageTimeout)
+	err := s.broadcastNow(ctx, addressee, true, peers...)
+	if err != nil {
+		s.logger.Debug("coalesced gossip flush failed", "addressee", addressee, "reason", reason, "batch_size", len(peers), "error", err)
+	}
+	cancel()
+}
+
+func (s *Service) recordCoalesceFlush(reason string, addressee swarm.Address, peers []swarm.Address) {
+	batchSize := len(peers)
+	if batchSize == 0 {
+		return
+	}
+
+	s.metrics.GossipCoalesceFlushTotal.WithLabelValues(reason).Inc()
+	s.metrics.GossipCoalesceFlushPeers.Add(float64(batchSize))
+	s.logger.Debug("coalesced gossip flush", "addressee", addressee, "reason", reason, "batch_size", batchSize)
+}
+
+func (s *Service) setCoalesceBufferGauge() {
+	s.metrics.GossipCoalesceBufferSize.Set(float64(s.gossipBuf.pendingAddressees()))
+}
+
 func (s *Service) startCheckPeersHandler() {
 	ctx, cancel := context.WithCancel(context.Background())
 	s.wg.Go(func() {