changes to peer mantainence logic

Author: aarshkshah1992

dht.go

@@ -109,7 +109,7 @@ type IpfsDHT struct {
 	enableProviders, enableValues bool
 
 	// maxLastSuccessfulOutboundThreshold is the max threshold/upper limit on the time duration
-	// between the current time and the last time a peer was useful to us.
+	// between the current time and the last time we successfully queried a peer.
 	maxLastSuccessfulOutboundThreshold float64
 
 	fixLowPeersChan chan struct{}
@@ -448,22 +448,37 @@ func (dht *IpfsDHT) putLocal(key string, rec *recpb.Record) error {
 
 // peerFound signals the routingTable that we've found a peer that
 // might support the DHT protocol.
+// If we have a connection a peer but no exchange of a query RPC ->
+//    LastQueriedAt=time.Now (so we don't ping it for some time for a liveliness check)
+//    LastUsefulAt=N/A
+// If we connect to a peer and exchange a query RPC ->
+//    LastQueriedAt=time.Now (same reason as above)
+//    LastUsefulAt=time.Now (so we give it some life in the RT without immediately evicting it)
+// If we query a peer we already have in our Routing Table ->
+//    LastQueriedAt=time.Now()
+//    LastUsefulAt remains unchanged
+// If we connect to a peer we already have in the RT but do not exchange a query (rare)
+//    Do Nothing.
 func (dht *IpfsDHT) peerFound(ctx context.Context, p peer.ID, queryPeer bool) {
 	logger.Debugw("peer found", "peer", p)
 	b, err := dht.validRTPeer(p)
 	if err != nil {
 		logger.Errorw("failed to validate if peer is a DHT peer", "peer", p, "error", err)
 	} else if b {
-		_, err := dht.routingTable.TryAddPeer(p, queryPeer)
+		newlyAdded, err := dht.routingTable.TryAddPeer(p, queryPeer)
 		if err != nil {
 			// peer not added.
 			return
 		}
 
-		// If we discovered the peer because of a query, we need to ensure we override the "zero" lastSuccessfulOutboundQuery
+		// If we freshly added the peer because of a query, we need to ensure we override the "zero" lastUsefulAt
 		// value that must have been set in the Routing Table for this peer when it was first added during a connection.
-		if queryPeer {
-			dht.routingTable.UpdateLastSuccessfulOutboundQuery(p, time.Now())
+		if newlyAdded && queryPeer {
+			dht.routingTable.UpdateLastUsefulAt(p, time.Now())
+		} else if queryPeer {
+			// the peer is already in our RT, but we just successfully queried it and so let's give it a
+			// bump on the query time so we don't ping it too soon for a liveliness check.
+			dht.routingTable.UpdateLastSuccessfulOutboundQueryAt(p, time.Now())
 		}
 	}
 }

dht_bootstrap.go

@@ -128,7 +128,7 @@ func (dht *IpfsDHT) startRefreshing() {
 			// ping Routing Table peers that haven't been hear of/from in the interval they should have been.
 			for _, ps := range dht.routingTable.GetPeerInfos() {
 				// ping the peer if it's due for a ping and evict it if the ping fails
-				if float64(time.Since(ps.LastSuccessfulOutboundQuery)) > dht.maxLastSuccessfulOutboundThreshold {
+				if float64(time.Since(ps.LastSuccessfulOutboundQueryAt)) > dht.maxLastSuccessfulOutboundThreshold {
 					livelinessCtx, cancel := context.WithTimeout(ctx, peerPingTimeout)
 					if err := dht.host.Connect(livelinessCtx, peer.AddrInfo{ID: ps.Id}); err != nil {
 						logger.Debugw("evicting peer after failed ping", "peer", ps.Id, "error", err)

go.mod

@@ -18,7 +18,7 @@ require (
 	github.com/libp2p/go-eventbus v0.1.0
 	github.com/libp2p/go-libp2p v0.7.4
 	github.com/libp2p/go-libp2p-core v0.5.1
-	github.com/libp2p/go-libp2p-kbucket v0.3.3
+	github.com/libp2p/go-libp2p-kbucket v0.3.4-0.20200409173621-ac6d9f25cfc5
 	github.com/libp2p/go-libp2p-peerstore v0.2.2
 	github.com/libp2p/go-libp2p-record v0.1.2
 	github.com/libp2p/go-libp2p-swarm v0.2.3

go.sum

@@ -232,6 +232,8 @@ github.com/libp2p/go-libp2p-discovery v0.3.0 h1:+JnYBRLzZQtRq0mK3xhyjBwHytLmJXMT
 github.com/libp2p/go-libp2p-discovery v0.3.0/go.mod h1:o03drFnz9BVAZdzC/QUQ+NeQOu38Fu7LJGEOK2gQltw=
 github.com/libp2p/go-libp2p-kbucket v0.3.3 h1:V2Zwv6QnCK6Who0iiJW2eUKwdlTYGJ2HnLViaolDOcs=
 github.com/libp2p/go-libp2p-kbucket v0.3.3/go.mod h1:IWFdYRBOYzaLEHnvrfzEkr+UcuveCXIoeO8QeFZSI6A=
+github.com/libp2p/go-libp2p-kbucket v0.3.4-0.20200409173621-ac6d9f25cfc5 h1:WKHaqIIF9mHebRjTtsOCGPY32Wr2TKOyKnF6KQ9LV8Q=
+github.com/libp2p/go-libp2p-kbucket v0.3.4-0.20200409173621-ac6d9f25cfc5/go.mod h1:sDQeCkD6yf/Yq8O+HPTNLyGa0TsJi+CURa6ELp1ppkk=
 github.com/libp2p/go-libp2p-loggables v0.1.0 h1:h3w8QFfCt2UJl/0/NW4K829HX/0S4KD31PQ7m8UXXO8=
 github.com/libp2p/go-libp2p-loggables v0.1.0/go.mod h1:EyumB2Y6PrYjr55Q3/tiJ/o3xoDasoRYM7nOzEpoa90=
 github.com/libp2p/go-libp2p-mplex v0.2.0/go.mod h1:Ejl9IyjvXJ0T9iqUTE1jpYATQ9NM3g+OtR+EMMODbKo=

query.go

@@ -174,7 +174,7 @@ func (dht *IpfsDHT) runQuery(ctx context.Context, target string, queryFn queryFn
 }
 
 func (q *query) recordPeerIsValuable(p peer.ID) {
-	q.dht.routingTable.UpdateLastSuccessfulOutboundQuery(p, time.Now())
+	q.dht.routingTable.UpdateLastUsefulAt(p, time.Now())
 }
 
 func (q *query) recordValuablePeers() {