feat: ping peers on routing table refresh

dht.go

@@ -26,6 +26,7 @@ import (
 	"github.com/libp2p/go-libp2p-kbucket/peerdiversity"
 	record "github.com/libp2p/go-libp2p-record"
 	recpb "github.com/libp2p/go-libp2p-record/pb"
+	"github.com/libp2p/go-libp2p/p2p/protocol/ping"
 
 	"github.com/gogo/protobuf/proto"
 	ds "github.com/ipfs/go-datastore"
@@ -68,6 +69,11 @@ const (
 	protectedBuckets = 2
 )
 
+const (
+	// MAGIC: timeout for receiving a pong from a peer in our routing table
+	pingTimeout = 5 * time.Second
+)
+
 type addPeerRTReq struct {
 	p         peer.ID
 	queryPeer bool
@@ -365,10 +371,27 @@ func makeRtRefreshManager(dht *IpfsDHT, cfg dhtcfg.Config, maxLastSuccessfulOutb
 		return err
 	}
 
+	pingFnc := func(ctx context.Context, p peer.ID) error {
+		timeoutCtx, cancel := context.WithTimeout(ctx, pingTimeout)
+		defer cancel()
+
+		// Just wait for a single ping
+		select {
+		case res, more := <-ping.Ping(timeoutCtx, dht.host, p):
+			if !more {
+				return timeoutCtx.Err()
+			}
+			return res.Error
+		case <-timeoutCtx.Done():
+			return timeoutCtx.Err()
+		}
+	}
+
 	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,6 +41,7 @@ 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.
+	refreshPingFnc      func(ctx context.Context, p peer.ID) error  // request to check liveness of remote peer
 	refreshQueryTimeout time.Duration                               // timeout for one refresh query
 
 	// interval between two periodic refreshes.
@@ -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)
 			}
 		}