fix(p2p): add exponential backoff for reacher (#5371)

Author: gacevicljubisa

pkg/p2p/libp2p/internal/reacher/metrics.go

@@ -0,0 +1,56 @@
+// 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 reacher
+
+import (
+	m "github.com/ethersphere/bee/v2/pkg/metrics"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+// metrics groups reacher related prometheus counters.
+type metrics struct {
+	Peers            prometheus.Gauge
+	PingAttemptCount prometheus.Counter
+	PingErrorCount   prometheus.Counter
+	PingDuration     prometheus.Histogram
+}
+
+// newMetrics is a convenient constructor for creating new metrics.
+func newMetrics() metrics {
+	const subsystem = "reacher"
+
+	return metrics{
+		Peers: prometheus.NewGauge(prometheus.GaugeOpts{
+			Namespace: m.Namespace,
+			Subsystem: subsystem,
+			Name:      "peers",
+			Help:      "Number of peers currently in the reacher queue.",
+		}),
+		PingAttemptCount: prometheus.NewCounter(prometheus.CounterOpts{
+			Namespace: m.Namespace,
+			Subsystem: subsystem,
+			Name:      "ping_attempt_count",
+			Help:      "Number of ping attempts.",
+		}),
+		PingErrorCount: prometheus.NewCounter(prometheus.CounterOpts{
+			Namespace: m.Namespace,
+			Subsystem: subsystem,
+			Name:      "ping_error_count",
+			Help:      "Number of failed ping attempts.",
+		}),
+		PingDuration: prometheus.NewHistogram(prometheus.HistogramOpts{
+			Namespace: m.Namespace,
+			Subsystem: subsystem,
+			Name:      "ping_duration_seconds",
+			Help:      "Ping latency distribution in seconds.",
+			Buckets:   []float64{.1, .25, .5, 1, 2, 5, 10, 15},
+		}),
+	}
+}
+
+// Metrics returns set of prometheus collectors.
+func (r *reacher) Metrics() []prometheus.Collector {
+	return m.PrometheusCollectorsFromFields(r.metrics)
+}

pkg/p2p/libp2p/internal/reacher/reacher.go

@@ -9,6 +9,7 @@ package reacher
 import (
 	"container/heap"
 	"context"
+	"math/rand/v2"
 	"sync"
 	"time"
 
@@ -19,16 +20,22 @@ import (
 )
 
 const (
-	pingTimeout        = time.Second * 15
-	workers            = 16
-	retryAfterDuration = time.Minute * 5
+	pingTimeout               = time.Second * 15
+	workers                   = 8
+	retryAfterDuration        = time.Minute * 5
+	maxFailBackoffExponent    = 4   // caps failure backoff at retryAfterDuration * 2^4 = 80 min
+	maxSuccessBackoffExponent = 2   // caps success backoff at retryAfterDuration * 2^2 = 20 min
+	jitterFactor              = 0.2 // ±20% randomization on retry intervals
 )
 
 type peer struct {
-	overlay    swarm.Address
-	addr       ma.Multiaddr
-	retryAfter time.Time
-	index      int // index in the heap
+	overlay      swarm.Address
+	addr         ma.Multiaddr
+	retryAfter   time.Time
+	failCount    int // consecutive ping failures for exponential backoff
+	successCount int // consecutive ping successes for exponential backoff
+	generation   int // incremented on reconnect; guards against stale notifyResult
+	index        int // index in the heap
 }
 
 type reacher struct {
@@ -44,6 +51,7 @@ type reacher struct {
 
 	wg sync.WaitGroup
 
+	metrics metrics
 	options *Options
 	logger  log.Logger
 }
@@ -52,6 +60,7 @@ type Options struct {
 	PingTimeout        time.Duration
 	Workers            int
 	RetryAfterDuration time.Duration
+	JitterFactor       float64 // ±N% randomization on retry intervals; 0 disables jitter
 }
 
 func New(streamer p2p.Pinger, notifier p2p.ReachableNotifier, o *Options, log log.Logger) *reacher {
@@ -62,6 +71,7 @@ func New(streamer p2p.Pinger, notifier p2p.ReachableNotifier, o *Options, log lo
 		peerHeap:  make(peerHeap, 0),
 		peerIndex: make(map[string]*peer),
 		notifier:  notifier,
+		metrics:   newMetrics(),
 		logger:    log.WithName("reacher").Register(),
 	}
 
@@ -70,6 +80,7 @@ func New(streamer p2p.Pinger, notifier p2p.ReachableNotifier, o *Options, log lo
 			PingTimeout:        pingTimeout,
 			Workers:            workers,
 			RetryAfterDuration: retryAfterDuration,
+			JitterFactor:       jitterFactor,
 		}
 	}
 	r.options = o
@@ -95,7 +106,6 @@ func (r *reacher) manage() {
 	}
 
 	for {
-
 		p, ok, tryAfter := r.tryAcquirePeer()
 
 		// if no peer is returned,
@@ -136,15 +146,22 @@ func (r *reacher) ping(c chan peer, ctx context.Context) {
 	defer r.wg.Done()
 	for p := range c {
 		func() {
+			r.metrics.PingAttemptCount.Inc()
 			ctxt, cancel := context.WithTimeout(ctx, r.options.PingTimeout)
 			defer cancel()
+			start := time.Now()
 			rtt, err := r.pinger.Ping(ctxt, p.addr)
 			if err != nil {
+				r.metrics.PingDuration.Observe(time.Since(start).Seconds())
+				r.metrics.PingErrorCount.Inc()
 				r.logger.Debug("ping failed", "peer", p.overlay.String(), "addr", p.addr.String(), "error", err)
 				r.notifier.Reachable(p.overlay, p2p.ReachabilityStatusPrivate)
+				r.notifyResult(p.overlay, false, p.generation)
 			} else {
+				r.metrics.PingDuration.Observe(rtt.Seconds())
 				r.logger.Debug("ping succeeded", "peer", p.overlay.String(), "addr", p.addr.String(), "rtt", rtt)
 				r.notifier.Reachable(p.overlay, p2p.ReachabilityStatusPublic)
+				r.notifyResult(p.overlay, true, p.generation)
 			}
 		}()
 	}
@@ -168,8 +185,10 @@ func (r *reacher) tryAcquirePeer() (peer, bool, time.Duration) {
 		return peer{}, false, time.Until(p.retryAfter)
 	}
 
-	// Update retryAfter and fix heap position
-	p.retryAfter = time.Now().Add(r.options.RetryAfterDuration)
+	// Set a temporary far-future retryAfter to prevent the manage loop from
+	// re-dispatching this peer while the ping is in flight. The actual
+	// retryAfter will be set by notifyResult after the ping completes.
+	p.retryAfter = now.Add(time.Hour)
 	heap.Fix(&r.peerHeap, p.index)
 
 	// Return a copy so callers can read fields without holding the lock.
@@ -190,13 +209,57 @@ func (r *reacher) Connected(overlay swarm.Address, addr ma.Multiaddr) {
 	if existing, ok := r.peerIndex[key]; ok {
 		existing.addr = addr              // Update address for reconnecting peer
 		existing.retryAfter = time.Time{} // Reset to trigger immediate re-ping
+		existing.failCount = 0            // Fresh start on reconnect
+		existing.successCount = 0         // Fresh start on reconnect
+		existing.generation++             // invalidate any in-flight notifyResult
 		heap.Fix(&r.peerHeap, existing.index)
 	} else {
 		p := &peer{overlay: overlay, addr: addr}
 		r.peerIndex[key] = p
 		heap.Push(&r.peerHeap, p)
+		r.metrics.Peers.Inc()
+	}
+
+	select {
+	case r.newPeer <- struct{}{}:
+	default:
+	}
+}
+
+// notifyResult updates the peer's retry schedule based on the ping outcome.
+// Both success and failure use exponential backoff with different caps:
+//   - Success: 5m → 10m → 20m (capped at 2^2), resets failCount
+//   - Failure: 5m → 10m → 20m → 40m → 80m (capped at 2^4), resets successCount
+//
+// The gen parameter is the generation captured when the ping was dispatched.
+// If the peer was reconnected (generation incremented) while the ping was
+// in flight, the stale result is discarded.
+func (r *reacher) notifyResult(overlay swarm.Address, success bool, gen int) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	p, ok := r.peerIndex[overlay.ByteString()]
+	if !ok {
+		return // peer was disconnected while ping was in flight
+	}
+	if p.generation != gen {
+		return // peer was reconnected; discard stale result
+	}
+
+	if success {
+		p.failCount = 0
+		p.successCount++
+		backoff := min(p.successCount, maxSuccessBackoffExponent)
+		p.retryAfter = time.Now().Add(r.jitter(r.options.RetryAfterDuration * time.Duration(1<<backoff)))
+	} else {
+		p.successCount = 0
+		p.failCount++
+		backoff := min(p.failCount, maxFailBackoffExponent)
+		p.retryAfter = time.Now().Add(r.jitter(r.options.RetryAfterDuration * time.Duration(1<<backoff)))
 	}
+	heap.Fix(&r.peerHeap, p.index)
 
+	// Wake the manage loop so it recalculates the next retry time.
 	select {
 	case r.newPeer <- struct{}{}:
 	default:
@@ -212,7 +275,19 @@ func (r *reacher) Disconnected(overlay swarm.Address) {
 	if p, ok := r.peerIndex[key]; ok {
 		heap.Remove(&r.peerHeap, p.index)
 		delete(r.peerIndex, key)
+		r.metrics.Peers.Dec()
+	}
+}
+
+// jitter adds ±JitterFactor randomization to a duration to prevent peers from
+// synchronizing their retry times and causing burst traffic.
+func (r *reacher) jitter(d time.Duration) time.Duration {
+	if r.options.JitterFactor == 0 {
+		return d
 	}
+	// rand.Float64() returns [0.0, 1.0), scale to [-JitterFactor, +JitterFactor)
+	j := 1.0 + r.options.JitterFactor*(2*rand.Float64()-1)
+	return time.Duration(float64(d) * j)
 }
 
 // Close stops the worker. Must be called once.