kademlia: wait on peer before retrying after connect err (#1449)

pkg/kademlia/kademlia.go

@@ -309,6 +309,14 @@ func (k *Kad) manage() {
 								}
 								k.logger.Debugf("peer not reachable from kademlia %s: %v", bzzAddr.String(), err)
 								k.logger.Warningf("peer not reachable when attempting to connect")
+
+								k.waitNextMu.Lock()
+								if _, ok := k.waitNext[peer.String()]; !ok {
+									// don't override existing data in the map
+									k.waitNext[peer.String()] = retryInfo{tryAfter: time.Now().Add(timeToRetry)}
+								}
+								k.waitNextMu.Unlock()
+
 								// continue to next
 								return nil
 							}
@@ -382,6 +390,14 @@ func (k *Kad) manage() {
 					}
 					k.logger.Debugf("peer not reachable from kademlia %s: %v", bzzAddr.String(), err)
 					k.logger.Warningf("peer not reachable when attempting to connect")
+
+					k.waitNextMu.Lock()
+					if _, ok := k.waitNext[peer.String()]; !ok {
+						// don't override existing data in the map
+						k.waitNext[peer.String()] = retryInfo{tryAfter: time.Now().Add(timeToRetry)}
+					}
+					k.waitNextMu.Unlock()
+
 					// continue to next
 					return false, false, nil
 				}

pkg/p2p/libp2p/libp2p.go

@@ -496,6 +496,7 @@ func (s *Service) Connect(ctx context.Context, addr ma.Multiaddr) (address *bzz.
 
 	if err := s.connectionBreaker.Execute(func() error { return s.host.Connect(ctx, *info) }); err != nil {
 		if errors.Is(err, breaker.ErrClosed) {
+			s.metrics.ConnectBreakerCount.Inc()
 			return nil, p2p.NewConnectionBackoffError(err, s.connectionBreaker.ClosedUntil())
 		}
 		return nil, err

pkg/p2p/libp2p/metrics.go

@@ -20,6 +20,7 @@ type metrics struct {
 	BlocklistedPeerCount    prometheus.Counter
 	BlocklistedPeerErrCount prometheus.Counter
 	DisconnectCount         prometheus.Counter
+	ConnectBreakerCount     prometheus.Counter
 }
 
 func newMetrics() metrics {
@@ -68,6 +69,12 @@ func newMetrics() metrics {
 			Name:      "disconnect_count",
 			Help:      "Number of peers we've disconnected from (initiated locally).",
 		}),
+		ConnectBreakerCount: prometheus.NewCounter(prometheus.CounterOpts{
+			Namespace: m.Namespace,
+			Subsystem: subsystem,
+			Name:      "connect_breaker_count",
+			Help:      "Number of times we got a closed breaker while connecting to another peer.",
+		}),
 	}
 }