Merge branch 'master' into provider-stats

Author: guillaumemichel

dht.go

@@ -746,39 +746,39 @@ func (dht *IpfsDHT) FindLocal(ctx context.Context, id peer.ID) peer.AddrInfo {
 	return peer.AddrInfo{}
 }
 
-// nearestPeersToQuery returns the routing tables closest peers.
-func (dht *IpfsDHT) nearestPeersToQuery(pmes *pb.Message, count int) []peer.ID {
-	closer := dht.routingTable.NearestPeers(kb.ConvertKey(string(pmes.GetKey())), count)
-	return closer
-}
-
-// betterPeersToQuery returns nearestPeersToQuery with some additional filtering
-func (dht *IpfsDHT) betterPeersToQuery(pmes *pb.Message, from peer.ID, count int) []peer.ID {
-	closer := dht.nearestPeersToQuery(pmes, count)
-
-	// no node? nil
-	if closer == nil {
+// closestPeersToQuery returns the closest peers to the target key from the
+// local routing table, filtering out self and the requester's peer ID.
+//
+// Per the IPFS Kademlia DHT spec, servers SHOULD NOT include themselves or the
+// requester in FIND_NODE responses (except for FIND_PEER when the target is
+// self or the requester, which is handled separately in handleFindPeer).
+func (dht *IpfsDHT) closestPeersToQuery(pmes *pb.Message, from peer.ID, count int) []peer.ID {
+	// Get count+1 closest peers to target key, so that we can filter out 'from'
+	// (and potentially 'self' if it somehow appears) and still return 'count' peers.
+	closestPeers := dht.routingTable.NearestPeers(kb.ConvertKey(string(pmes.GetKey())), count+1)
+
+	if len(closestPeers) == 0 {
 		logger.Infow("no closer peers to send", from)
 		return nil
 	}
 
-	filtered := make([]peer.ID, 0, len(closer))
-	for _, clp := range closer {
-
-		// == to self? thats bad
-		if clp == dht.self {
-			logger.Error("BUG betterPeersToQuery: attempted to return self! this shouldn't happen...")
-			return nil
+	filtered := make([]peer.ID, 0, min(len(closestPeers), count))
+	for _, p := range closestPeers {
+		// Per spec: don't include self in responses. This should never happen since
+		// self should not be in the routing table, but check defensively.
+		if p == dht.self {
+			logger.Debugw("self found in routing table, skipping", "key", string(pmes.GetKey()))
+			continue
 		}
-		// Dont send a peer back themselves
-		if clp == from {
+		// Per spec: don't include requester in responses (exception handled in handleFindPeer).
+		if p == from {
 			continue
 		}
-
-		filtered = append(filtered, clp)
+		filtered = append(filtered, p)
+		if len(filtered) >= count {
+			break
+		}
 	}
-
-	// ok seems like closer nodes
 	return filtered
 }
 

handlers.go

@@ -7,7 +7,7 @@ import (
 	"time"
 
 	"github.com/libp2p/go-libp2p/core/peer"
-	pstore "github.com/libp2p/go-libp2p/p2p/host/peerstore"
+	"github.com/libp2p/go-libp2p/core/peerstore"
 
 	ds "github.com/ipfs/go-datastore"
 	"github.com/libp2p/go-libp2p-kad-dht/internal"
@@ -66,12 +66,11 @@ func (dht *IpfsDHT) handleGetValue(ctx context.Context, p peer.ID, pmes *pb.Mess
 	resp.Record = rec
 
 	// Find closest peer on given cluster to desired key and reply with that info
-	closer := dht.betterPeersToQuery(pmes, p, dht.bucketSize)
-	if len(closer) > 0 {
-		// TODO: pstore.PeerInfos should move to core (=> peerstore.AddrInfos).
-		closerinfos := pstore.PeerInfos(dht.peerstore, closer)
-		for _, pi := range closerinfos {
-			logger.Debugf("handleGetValue returning closer peer: '%s'", pi.ID)
+	closestPeers := dht.closestPeersToQuery(pmes, p, dht.bucketSize)
+	if len(closestPeers) > 0 {
+		closestInfos := peerstore.AddrInfos(dht.peerstore, closestPeers)
+		for _, pi := range closestInfos {
+			logger.Debugf("handleGetValue returning closest peer: '%s'", pi.ID)
 			if len(pi.Addrs) < 1 {
 				logger.Warnw("no addresses on peer being sent",
 					"local", dht.self,
@@ -81,7 +80,7 @@ func (dht *IpfsDHT) handleGetValue(ctx context.Context, p peer.ID, pmes *pb.Mess
 			}
 		}
 
-		resp.CloserPeers = pb.PeerInfosToPBPeers(dht.host.Network(), closerinfos)
+		resp.CloserPeers = pb.PeerInfosToPBPeers(dht.host.Network(), closestInfos)
 	}
 
 	return resp, nil
@@ -252,42 +251,26 @@ func (dht *IpfsDHT) handlePing(_ context.Context, p peer.ID, pmes *pb.Message) (
 
 func (dht *IpfsDHT) handleFindPeer(ctx context.Context, from peer.ID, pmes *pb.Message) (_ *pb.Message, _err error) {
 	resp := pb.NewMessage(pmes.GetType(), nil, pmes.GetClusterLevel())
-	var closest []peer.ID
 
 	if len(pmes.GetKey()) == 0 {
 		return nil, errors.New("handleFindPeer with empty key")
 	}
 
-	// if looking for self... special case where we send it on CloserPeers.
 	targetPid := peer.ID(pmes.GetKey())
-	closest = dht.betterPeersToQuery(pmes, from, dht.bucketSize)
-
-	// Never tell a peer about itself.
-	if targetPid != from {
-		// Add the target peer to the set of closest peers if
-		// not already present in our routing table.
-		//
-		// Later, when we lookup known addresses for all peers
-		// in this set, we'll prune this peer if we don't
-		// _actually_ know where it is.
-		found := false
-		for _, p := range closest {
-			if targetPid == p {
-				found = true
-				break
-			}
-		}
-		if !found {
-			closest = append(closest, targetPid)
-		}
-	}
+	closest := dht.closestPeersToQuery(pmes, from, dht.bucketSize)
 
-	if closest == nil {
-		return resp, nil
+	// Prepend targetPid to the front of the list if not already present.
+	// targetPid is always the closest key to itself.
+	//
+	// Per IPFS Kademlia DHT spec: FIND_PEER has a special exception where the
+	// target peer MUST be included in the response (if present in peerstore),
+	// even if it is self, the requester, or not a DHT server. This allows peers
+	// to discover multiaddresses for any peer, not just DHT servers.
+	if len(closest) == 0 || closest[0] != targetPid {
+		closest = append([]peer.ID{targetPid}, closest...)
 	}
 
-	// TODO: pstore.PeerInfos should move to core (=> peerstore.AddrInfos).
-	closestinfos := pstore.PeerInfos(dht.peerstore, closest)
+	closestinfos := peerstore.AddrInfos(dht.peerstore, closest)
 	// possibly an over-allocation but this array is temporary anyways.
 	withAddresses := make([]peer.AddrInfo, 0, len(closestinfos))
 	for _, pi := range closestinfos {
@@ -326,11 +309,10 @@ func (dht *IpfsDHT) handleGetProviders(ctx context.Context, p peer.ID, pmes *pb.
 
 	resp.ProviderPeers = pb.PeerInfosToPBPeers(dht.host.Network(), filtered)
 
-	// Also send closer peers.
-	closer := dht.betterPeersToQuery(pmes, p, dht.bucketSize)
-	if closer != nil {
-		// TODO: pstore.PeerInfos should move to core (=> peerstore.AddrInfos).
-		infos := pstore.PeerInfos(dht.peerstore, closer)
+	// Also send closest dht servers we know about.
+	closestPeers := dht.closestPeersToQuery(pmes, p, dht.bucketSize)
+	if closestPeers != nil {
+		infos := peerstore.AddrInfos(dht.peerstore, closestPeers)
 		resp.CloserPeers = pb.PeerInfosToPBPeers(dht.host.Network(), infos)
 	}
 

provider/provider.go

@@ -573,6 +573,7 @@ func (s *SweepingProvider) getAvgPrefixLenNoLock() (int, error) {
 // (typically 1 or 2), because exploring the keyspace would add too much
 // overhead for a small number of keys.
 func (s *SweepingProvider) vanillaProvide(k mh.Multihash, reprovide bool) (bitstr.Key, error) {
+	keys := []mh.Multihash{k}
 	// Add provider record to local provider store.
 	s.addLocalRecord(k)
 	// Get peers to which the record will be allocated.
@@ -584,7 +585,7 @@ func (s *SweepingProvider) vanillaProvide(k mh.Multihash, reprovide bool) (bitst
 	addrInfo := peer.AddrInfo{ID: s.peerid, Addrs: s.getSelfAddrs()}
 	keysAllocations := make(map[peer.ID][]mh.Multihash)
 	for _, p := range peers {
-		keysAllocations[p] = []mh.Multihash{k}
+		keysAllocations[p] = keys
 	}
 	reachablePeers, err := s.sendProviderRecords(keysAllocations, addrInfo, 1)
 
@@ -598,6 +599,10 @@ func (s *SweepingProvider) vanillaProvide(k mh.Multihash, reprovide bool) (bitst
 	}
 	s.stats.cycleStatsLk.Unlock()
 
+	if err == nil {
+		logger.Debugw("sent provider record", "prefix", coveredPrefix, "count", 1, "keys", keys)
+	}
+
 	return coveredPrefix, err
 }
 
@@ -1160,6 +1165,7 @@ func (s *SweepingProvider) batchProvide(prefix bitstr.Key, keys []mh.Multihash)
 	if keyCount == 0 {
 		return
 	}
+	logger.Debugw("batchProvide called", "prefix", prefix, "count", len(keys))
 	addrInfo, ok := s.selfAddrInfo()
 	if !ok {
 		// Don't provide if the node doesn't have a valid address to include in the
@@ -1415,6 +1421,9 @@ func (s *SweepingProvider) provideRegions(regions []keyspace.Region, addrInfo pe
 			}
 			continue
 		}
+		keyCount := len(allKeys)
+		s.provideCounter.Add(s.ctx, int64(keyCount))
+		logger.Debugw("sent provider records", "prefix", r.Prefix, "count", keyCount, "keys", allKeys)
 	}
 	// If at least 1 regions was provided, we don't consider it a failure.
 	return errCount < len(regions)
@@ -1506,25 +1515,6 @@ func (s *SweepingProvider) Clear() int {
 	return s.provideQueue.Clear()
 }
 
-// ProvideState encodes the current relationship between this node and `key`.
-type ProvideState uint8
-
-const (
-	StateUnknown  ProvideState = iota // we have no record of the key
-	StateQueued                       // key is queued to be provided
-	StateProvided                     // key was provided at least once
-)
-
-// ProvideStatus reports the provider’s view of a key.
-//
-// When `state == StateProvided`, `lastProvide` is the wall‑clock time of the
-// most recent successful provide operation (UTC).
-// For `StateQueued` or `StateUnknown`, `lastProvide` is the zero `time.Time`.
-func (s *SweepingProvider) ProvideStatus(key mh.Multihash) (state ProvideState, lastProvide time.Time) {
-	// TODO: implement me
-	return StateUnknown, time.Time{}
-}
-
 // AddToSchedule makes sure the prefixes associated with the supplied keys are
 // scheduled to be reprovided.
 //