feat: ping peers on routing table refresh

We have seen in the past that there are peers in the IPFS DHT that let you connect to them but then refuse to speak any protocol. This was mainly due to resource manager killing the connection if limits were exceeded. We have seen that such peers are pushed to the edge of the DHT - meaning, they get already pruned from lower buckets. However, they won't get pruned from higher ones, because we only try to connect to them and not speak anything on that connection.

This change adds a ping message to the liveness check on routing table refreshes.

dht.go

@@ -365,10 +365,15 @@ func makeRtRefreshManager(dht *IpfsDHT, cfg dhtcfg.Config, maxLastSuccessfulOutb
 		return err
 	}
 
+	pingFnc := func(ctx context.Context, p peer.ID) error {
+		return dht.protoMessenger.Ping(ctx, p)
+	}
+
 	r, err := rtrefresh.NewRtRefreshManager(
 		dht.host, dht.routingTable, cfg.RoutingTable.AutoRefresh,
 		keyGenFnc,
 		queryFnc,
+		pingFnc,
 		cfg.RoutingTable.RefreshQueryTimeout,
 		cfg.RoutingTable.RefreshInterval,
 		maxLastSuccessfulOutboundThreshold,

rtrefresh/rt_refresh_manager.go

@@ -41,7 +41,8 @@ type RtRefreshManager struct {
 	enableAutoRefresh   bool                                        // should run periodic refreshes ?
 	refreshKeyGenFnc    func(cpl uint) (string, error)              // generate the key for the query to refresh this cpl
 	refreshQueryFnc     func(ctx context.Context, key string) error // query to run for a refresh.
-	refreshQueryTimeout time.Duration                               // timeout for one refresh query
+	refreshPingFnc      func(ctx context.Context, p peer.ID) error
+	refreshQueryTimeout time.Duration // timeout for one refresh query
 
 	// interval between two periodic refreshes.
 	// also, a cpl wont be refreshed if the time since it was last refreshed
@@ -57,6 +58,7 @@ type RtRefreshManager struct {
 func NewRtRefreshManager(h host.Host, rt *kbucket.RoutingTable, autoRefresh bool,
 	refreshKeyGenFnc func(cpl uint) (string, error),
 	refreshQueryFnc func(ctx context.Context, key string) error,
+	refreshPingFnc func(ctx context.Context, p peer.ID) error,
 	refreshQueryTimeout time.Duration,
 	refreshInterval time.Duration,
 	successfulOutboundQueryGracePeriod time.Duration,
@@ -73,6 +75,7 @@ func NewRtRefreshManager(h host.Host, rt *kbucket.RoutingTable, autoRefresh bool
 		enableAutoRefresh: autoRefresh,
 		refreshKeyGenFnc:  refreshKeyGenFnc,
 		refreshQueryFnc:   refreshQueryFnc,
+		refreshPingFnc:    refreshPingFnc,
 
 		refreshQueryTimeout:                refreshQueryTimeout,
 		refreshInterval:                    refreshInterval,
@@ -178,12 +181,21 @@ func (r *RtRefreshManager) loop() {
 				wg.Add(1)
 				go func(ps kbucket.PeerInfo) {
 					defer wg.Done()
+
 					livelinessCtx, cancel := context.WithTimeout(r.ctx, peerPingTimeout)
+					defer cancel()
+
 					if err := r.h.Connect(livelinessCtx, peer.AddrInfo{ID: ps.Id}); err != nil {
+						logger.Debugw("evicting peer after failed connection", "peer", ps.Id, "error", err)
+						r.rt.RemovePeer(ps.Id)
+						return
+					}
+
+					if err := r.refreshPingFnc(livelinessCtx, ps.Id); err != nil {
 						logger.Debugw("evicting peer after failed ping", "peer", ps.Id, "error", err)
 						r.rt.RemovePeer(ps.Id)
+						return
 					}
-					cancel()
 				}(ps)
 			}
 		}