feat(discovery): add jitter and fix backoff

share/p2p/discovery/discovery.go

@@ -4,16 +4,18 @@
 	"context"
 	"errors"
 	"fmt"
+	"math/rand"
 	"time"
 
+	"golang.org/x/sync/errgroup"
+
 	logging "github.com/ipfs/go-log/v2"
 	"github.com/libp2p/go-libp2p/core/discovery"
 	"github.com/libp2p/go-libp2p/core/event"
 	"github.com/libp2p/go-libp2p/core/host"
 	"github.com/libp2p/go-libp2p/core/network"
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/libp2p/go-libp2p/p2p/host/eventbus"
-	"golang.org/x/sync/errgroup"
 )
 
 var log = logging.Logger("share/discovery")
@@ -33,12 +35,12 @@
 
 	// logInterval defines the time interval at which a warning message will be logged
 	// if the desired number of nodes is not detected.
-	logInterval = 5 * time.Minute
+	logInterval            = 5 * time.Minute
+	maxBackoffDuration     = 10 * time.Minute
+	backoffInitialDuration = 5 * time.Second
+	jitterFactor           = 0.1
 )
 
-// discoveryRetryTimeout defines time interval between discovery attempts, needed for tests
-var discoveryRetryTimeout = retryTimeout
-
 // Discovery combines advertise and discover services and allows to store discovered nodes.
 // TODO: The code here gets horribly hairy, so we should refactor this at some point
 type Discovery struct {
@@ -57,7 +59,8 @@
 
 	cancel context.CancelFunc
 
-	params *Parameters
+	params          *Parameters
+	backoffDuration time.Duration
 }
 
 type OnUpdatedPeers func(peerID peer.ID, isAdded bool)
@@ -86,14 +89,15 @@
 	}
 	o := newOptions(opts...)
 	return &Discovery{
-		tag:            tag,
-		set:            newLimitedSet(params.PeersLimit),
-		host:           h,
-		disc:           d,
-		connector:      newBackoffConnector(h, defaultBackoffFactory),
-		onUpdatedPeers: o.onUpdatedPeers,
-		params:         params,
-		triggerDisc:    make(chan struct{}),
+		tag:             tag,
+		set:             newLimitedSet(params.PeersLimit),
+		host:            h,
+		disc:            d,
+		connector:       newBackoffConnector(h, defaultBackoffFactory),
+		onUpdatedPeers:  o.onUpdatedPeers,
+		params:          params,
+		triggerDisc:     make(chan struct{}),
+		backoffDuration: backoffInitialDuration,
 	}, nil
 }
 
@@ -199,21 +203,25 @@
 // It initiates peer discovery upon request and restarts the process until the soft limit is
 // reached.
 func (d *Discovery) discoveryLoop(ctx context.Context) {
-	t := time.NewTicker(discoveryRetryTimeout)
-	defer t.Stop()
+	backoffTimer := time.NewTimer(0)
+	defer backoffTimer.Stop()
 
 	warnTicker := time.NewTicker(logInterval)
 	defer warnTicker.Stop()
 
 	for {
-		// drain all previous ticks from the channel
-		drainChannel(t.C)
 		select {
-		case <-t.C:
+		case <-backoffTimer.C:
 			if !d.discover(ctx) {
-				// rerun discovery if the number of peers hasn't reached the limit
+				d.backoffDuration += backoffInitialDuration
+				if d.backoffDuration > maxBackoffDuration {
+					d.backoffDuration = maxBackoffDuration
+				}
+				jitter := time.Duration(float64(d.backoffDuration) * jitterFactor * (rand.Float64()*2 - 1))
+				backoffTimer.Reset(d.backoffDuration + jitter)
 				continue
 			}
+			d.backoffDuration = backoffInitialDuration
 		case <-warnTicker.C:
 			if d.set.Size() < d.set.Limit() {
 				log.Warnf(
@@ -222,8 +230,6 @@
 					d.tag, logInterval, d.set.Size(), d.set.Limit(),
 				)
 			}
-			// Do not break the loop; just continue
-			continue
 		case <-ctx.Done():
 			return
 		}
@@ -375,13 +381,3 @@
 	d.host.ConnManager().Protect(peer.ID, d.tag)
 	return true
 }
-
-func drainChannel(c <-chan time.Time) {
-	for {
-		select {
-		case <-c:
-		default:
-			return
-		}
-	}
-}

share/p2p/discovery/discovery_test.go

@@ -26,9 +26,7 @@ const (
 func TestDiscovery(t *testing.T) {
 	const nodes = 10 // higher number brings higher coverage
 
-	discoveryRetryTimeout = time.Millisecond * 100 // defined in discovery.go
-
-	ctx, cancel := context.WithTimeout(context.Background(), time.Second*20)
+	ctx, cancel := context.WithTimeout(context.Background(), time.Minute*10)
 	t.Cleanup(cancel)
 
 	tn := newTestnet(ctx, t)