fix(share/discovery)!: revamp Discovery

nodebuilder/tests/p2p_test.go

@@ -215,6 +215,5 @@ func TestRestartNodeDiscovery(t *testing.T) {
 
 func setTimeInterval(cfg *nodebuilder.Config, interval time.Duration) {
 	cfg.P2P.RoutingTableRefreshPeriod = interval
-	cfg.Share.Discovery.DiscoveryRetryTimeout = interval
 	cfg.Share.Discovery.AdvertiseInterval = interval
 }

share/availability/discovery/backoff.go

@@ -85,10 +85,7 @@ func (b *backoffConnector) HasBackoff(p peer.ID) bool {
 	b.cacheLk.Lock()
 	cache, ok := b.cacheData[p]
 	b.cacheLk.Unlock()
-	if ok && time.Now().Before(cache.nexttry) {
-		return true
-	}
-	return false
+	return ok && time.Now().Before(cache.nexttry)
 }
 
 // GC is a perpetual GCing loop.

share/availability/discovery/discovery.go

@@ -23,6 +23,8 @@ const (
 	// events in libp2p
 	eventbusBufSize = 32
 
+	// findPeersStuckWarnDelay is the duration after which findPeers will log an error message to
+	// notify that it is stuck.
 	findPeersStuckWarnDelay = time.Minute
 )
 
@@ -35,20 +37,20 @@ type Parameters struct {
 	// PeersLimit defines the soft limit of FNs to connect to via discovery.
 	// Set 0 to disable.
 	PeersLimit uint
-	// DiscoveryRetryTimeout is an interval between discovery attempts
-	// when we discovered lower than PeersLimit peers.
-	// Set -1 to disable.
-	DiscoveryRetryTimeout time.Duration
 	// AdvertiseInterval is a interval between advertising sessions.
 	// Set -1 to disable.
 	// NOTE: only full and bridge can advertise themselves.
 	AdvertiseInterval time.Duration
+	// discoveryRetryTimeout is an interval between discovery attempts
+	// when we discovered lower than PeersLimit peers.
+	// Set -1 to disable.
+	discoveryRetryTimeout time.Duration
 }
 
 func DefaultParameters() Parameters {
 	return Parameters{
 		PeersLimit:            5,
-		DiscoveryRetryTimeout: time.Second * 1,
+		discoveryRetryTimeout: time.Second,
 		AdvertiseInterval:     time.Hour * 8,
 	}
 }
@@ -68,7 +70,7 @@ type Discovery struct {
 	connector      *backoffConnector
 	onUpdatedPeers OnUpdatedPeers
 
-	triggerDisq chan struct{}
+	triggerDisc chan struct{}
 
 	cancel context.CancelFunc
 }
@@ -88,7 +90,7 @@ func NewDiscovery(
 		disc:           d,
 		connector:      newBackoffConnector(h, defaultBackoffFactory),
 		onUpdatedPeers: func(peer.ID, bool) {},
-		triggerDisq:    make(chan struct{}),
+		triggerDisc:    make(chan struct{}),
 	}
 }
 
@@ -115,7 +117,7 @@ func (d *Discovery) WithOnPeersUpdate(f OnUpdatedPeers) {
 
 func (d *Discovery) triggerDiscovery() {
 	select {
-	case d.triggerDisq <- struct{}{}:
+	case d.triggerDisc <- struct{}{}:
 	default:
 	}
 }
@@ -153,11 +155,11 @@ func (d *Discovery) handlePeerFound(ctx context.Context, peer peer.AddrInfo) boo
 	}
 
 	if !d.set.Add(peer.ID) {
-		log.Debugw("peer is already in discovery set", "peer", peer.ID)
+		log.Debug("peer is already in discovery set")
 		return false
 	}
 	d.onUpdatedPeers(peer.ID, true)
-	log.Debugw("added peer to set", "peer", peer.ID)
+	log.Debug("added peer to set")
 
 	// tag to protect peer from being killed by ConnManager
 	// NOTE: This is does not protect from remote killing the connection.
@@ -220,7 +222,7 @@ func (d *Discovery) ensurePeers(ctx context.Context) {
 	}()
 	go d.connector.GC(ctx)
 
-	t := time.NewTicker(d.params.DiscoveryRetryTimeout)
+	t := time.NewTicker(d.params.discoveryRetryTimeout)
 	defer t.Stop()
 	for {
 		// drain all previous ticks from channel
@@ -237,7 +239,7 @@ func (d *Discovery) ensurePeers(ctx context.Context) {
 		}
 
 		select {
-		case <-d.triggerDisq:
+		case <-d.triggerDisc:
 		case <-ctx.Done():
 			return
 		}

share/availability/discovery/discovery_test.go

@@ -26,7 +26,7 @@ func TestDiscovery(t *testing.T) {
 
 	peerA := tn.discovery(Parameters{
 		PeersLimit:            nodes,
-		DiscoveryRetryTimeout: time.Millisecond * 100,
+		discoveryRetryTimeout: time.Millisecond * 100,
 		AdvertiseInterval:     -1, // we don't want to be found but only find
 	})
 
@@ -43,7 +43,7 @@ func TestDiscovery(t *testing.T) {
 	for i := range discs {
 		discs[i] = tn.discovery(Parameters{
 			PeersLimit:            0,
-			DiscoveryRetryTimeout: -1,
+			discoveryRetryTimeout: -1,
 			AdvertiseInterval:     time.Millisecond * 100,
 		})
 

share/availability/full/testing.go

@@ -37,12 +37,10 @@ func Node(dn *availability_test.TestDagNet) *availability_test.TestNode {
 }
 
 func TestAvailability(getter share.Getter) *ShareAvailability {
-	disc := discovery.NewDiscovery(nil, routing.NewRoutingDiscovery(
-		routinghelpers.Null{}), discovery.Parameters{
-		PeersLimit:            10,
-		DiscoveryRetryTimeout: time.Second,
-		AdvertiseInterval:     time.Second,
-	},
-	)
+	params := discovery.DefaultParameters()
+	params.AdvertiseInterval = time.Second
+	params.PeersLimit = 10
+	disc := discovery.NewDiscovery(nil,
+		routing.NewRoutingDiscovery(routinghelpers.Null{}), params)
 	return NewShareAvailability(nil, getter, disc)
 }

share/getters/shrex_test.go

@@ -159,9 +159,8 @@ func testManager(ctx context.Context, host host.Host, headerSub libhead.Subscrib
 
 	disc := discovery.NewDiscovery(nil,
 		routingdisc.NewRoutingDiscovery(routinghelpers.Null{}), discovery.Parameters{
-			PeersLimit:            10,
-			DiscoveryRetryTimeout: time.Second,
-			AdvertiseInterval:     time.Second,
+			PeersLimit:        10,
+			AdvertiseInterval: time.Second,
 		},
 	)
 	connGater, err := conngater.NewBasicConnectionGater(ds_sync.MutexWrap(datastore.NewMapDatastore()))

share/p2p/peers/manager_test.go

@@ -394,9 +394,8 @@ func TestIntegration(t *testing.T) {
 			nw.Hosts()[0],
 			routingdisc.NewRoutingDiscovery(router1),
 			discovery.Parameters{
-				PeersLimit:            0,
-				DiscoveryRetryTimeout: time.Second,
-				AdvertiseInterval:     time.Second,
+				PeersLimit:        0,
+				AdvertiseInterval: time.Second,
 			},
 		)
 
@@ -408,9 +407,8 @@ func TestIntegration(t *testing.T) {
 			nw.Hosts()[1],
 			routingdisc.NewRoutingDiscovery(router2),
 			discovery.Parameters{
-				PeersLimit:            10,
-				DiscoveryRetryTimeout: time.Second,
-				AdvertiseInterval:     time.Second,
+				PeersLimit:        10,
+				AdvertiseInterval: time.Second,
 			},
 		)
 		err = fnDisc.Start(ctx)
@@ -466,9 +464,8 @@ func testManager(ctx context.Context, headerSub libhead.Subscriber[*header.Exten
 
 	disc := discovery.NewDiscovery(nil,
 		routingdisc.NewRoutingDiscovery(routinghelpers.Null{}), discovery.Parameters{
-			PeersLimit:            0,
-			DiscoveryRetryTimeout: time.Second,
-			AdvertiseInterval:     time.Second,
+			PeersLimit:        0,
+			AdvertiseInterval: time.Second,
 		})
 	connGater, err := conngater.NewBasicConnectionGater(sync.MutexWrap(datastore.NewMapDatastore()))
 	if err != nil {