dht: do not err on context deadline (#6581)

Author: algorandskiy

network/p2p/capabilities.go

@@ -18,8 +18,8 @@ package p2p
 
 import (
 	"context"
+	"errors"
 	randv1 "math/rand"
-	"math/rand/v2"
 	"sync"
 	"time"
 
@@ -130,39 +130,15 @@ func (c *CapabilitiesDiscovery) AdvertiseCapabilities(capabilities ...Capability
 		eb := ebf()
 
 		for {
-			// shuffle capabilities to advertise in random order
-			// since the DHT's internal advertisement happens concurrently for peers in its routing table
-			// any peer error does not prevent advertisement of other peers.
-			// on repeated advertisement, we want to avoid the same order to make sure all capabilities are advertised.
-			if len(capabilities) > 1 {
-				rand.Shuffle(len(capabilities), func(i, j int) {
-					capabilities[i], capabilities[j] = capabilities[j], capabilities[i]
-				})
-			}
 			select {
 			case <-c.dht.Context().Done():
 				return
 			case <-nextExecution:
-				var err error
-				advertisementInterval := maxAdvertisementInterval
-				for _, capa := range capabilities {
-					ttl, err0 := c.advertise(c.dht.Context(), string(capa))
-					if err0 != nil {
-						err = err0
-						loggerFn := c.log.Errorf
-						if err0 == kbucket.ErrLookupFailure {
-							// No peers in a routing table, it is typical for startup and not an error
-							loggerFn = c.log.Debugf
-						}
-						loggerFn("failed to advertise for capability %s: %v", capa, err0)
-						break
-					}
-					if ttl < advertisementInterval {
-						advertisementInterval = ttl
-					}
-					c.log.Infof("advertised capability %s", capa)
+				advertisementInterval, err := c.advertiseCaps(capabilities)
+				if c.dht.Context().Err() != nil {
+					return
 				}
-				// If we failed to advertise, retry every according to exp jitter delays until successful
+				// If any capability failed to advertise, retry according to exp jitter delays
 				if err != nil {
 					nextExecution = time.After(eb.Delay())
 				} else {
@@ -175,6 +151,44 @@ func (c *CapabilitiesDiscovery) AdvertiseCapabilities(capabilities ...Capability
 	}()
 }
 
+// advertiseCaps advertises all capabilities concurrently and returns
+// the minimum re-advertisement interval and the first error encountered (if any)
+func (c *CapabilitiesDiscovery) advertiseCaps(capabilities []Capability) (time.Duration, error) {
+	type result struct {
+		capa Capability
+		ttl  time.Duration
+		err  error
+	}
+	results := make(chan result, len(capabilities))
+	for _, capa := range capabilities {
+		go func(cap Capability) {
+			ttl, err := c.advertise(c.dht.Context(), string(cap))
+			results <- result{capa: cap, ttl: ttl, err: err}
+		}(capa)
+	}
+
+	var aggErr error
+	advertisementInterval := maxAdvertisementInterval
+	for range capabilities {
+		r := <-results
+		if r.err != nil {
+			aggErr = errors.Join(aggErr, r.err)
+			loggerFn := c.log.Warnf
+			if r.err == kbucket.ErrLookupFailure {
+				// No peers in a routing table, it is typical for startup and not an error
+				loggerFn = c.log.Debugf
+			}
+			loggerFn("failed to advertise for capability %s: %v", r.capa, r.err)
+			continue
+		}
+		if r.ttl < advertisementInterval {
+			advertisementInterval = r.ttl
+		}
+		c.log.Infof("advertised capability %s", r.capa)
+	}
+	return advertisementInterval, aggErr
+}
+
 // Sizer exposes the Size method
 type Sizer interface {
 	Size() int

network/p2pNetwork_test.go

@@ -108,7 +108,7 @@ func TestP2PSubmitTX(t *testing.T) {
 	)
 	require.Eventually(t, func() bool {
 		return netA.hasPeers() && netB.hasPeers() && netC.hasPeers()
-	}, 5*time.Second, 50*time.Millisecond)
+	}, 10*time.Second, 50*time.Millisecond)
 
 	// for some reason the above check is not enough in race builds on CI
 	time.Sleep(time.Second) // give time for peers to connect.
@@ -203,7 +203,7 @@ func TestP2PSubmitTXNoGossip(t *testing.T) {
 
 	require.Eventually(t, func() bool {
 		return netA.hasPeers() && netB.hasPeers() && netC.hasPeers()
-	}, 5*time.Second, 50*time.Millisecond)
+	}, 10*time.Second, 50*time.Millisecond)
 
 	time.Sleep(time.Second) // give time for peers to connect.
 
@@ -288,7 +288,7 @@ func TestP2PSubmitWS(t *testing.T) {
 
 	require.Eventually(t, func() bool {
 		return netA.hasPeers() && netB.hasPeers() && netC.hasPeers()
-	}, 5*time.Second, 50*time.Millisecond)
+	}, 10*time.Second, 50*time.Millisecond)
 
 	time.Sleep(time.Second) // give time for peers to connect.
 
@@ -731,7 +731,7 @@ func TestP2PNetworkDHTCapabilities(t *testing.T) {
 
 			require.Eventually(t, func() bool {
 				return netA.hasPeers() && netB.hasPeers() && netC.hasPeers()
-			}, 2*time.Second, 50*time.Millisecond)
+			}, 10*time.Second, 50*time.Millisecond)
 
 			t.Logf("peers connected")
 
@@ -1009,7 +1009,7 @@ func TestP2PRelay(t *testing.T) {
 
 	require.Eventually(t, func() bool {
 		return netA.hasPeers() && netB.hasPeers()
-	}, 2*time.Second, 50*time.Millisecond)
+	}, 10*time.Second, 50*time.Millisecond)
 
 	type logMessages struct {
 		msgs [][]byte
@@ -1089,7 +1089,7 @@ func TestP2PRelay(t *testing.T) {
 	require.Eventually(t, func() bool {
 		return netA.hasPeers() && netB.hasPeers() && netC.hasPeers() &&
 			netA.hasPeer(netB.service.ID()) && netA.hasPeer(netC.service.ID())
-	}, 2*time.Second, 50*time.Millisecond)
+	}, 10*time.Second, 50*time.Millisecond)
 
 	// Wait for gossipsub heartbeat to establish mesh links.
 	// ListPeersForTopic returns subscribed peers but mesh links are established
@@ -1367,7 +1367,7 @@ func TestP2PwsStreamHandlerDedup(t *testing.T) {
 	// now allow the peer made outgoing connection to handle conn closing initiated by the other side
 	require.Eventually(t, func() bool {
 		return !netA.hasPeers() && !netB.hasPeers()
-	}, 2*time.Second, 50*time.Millisecond)
+	}, 10*time.Second, 50*time.Millisecond)
 }
 
 // TestP2PEnableGossipService_NodeDisable ensures that a node with EnableGossipService=false
@@ -1421,7 +1421,7 @@ func TestP2PEnableGossipService_NodeDisable(t *testing.T) {
 
 			require.Eventually(t, func() bool {
 				return netA.hasPeers() && netB.hasPeers()
-			}, 1*time.Second, 50*time.Millisecond)
+			}, 10*time.Second, 50*time.Millisecond)
 
 			testTag := protocol.AgreementVoteTag