double hashing implementation

dht.go

@@ -685,6 +685,12 @@ func (dht *IpfsDHT) FindLocal(id peer.ID) peer.AddrInfo {
 
 // nearestPeersToQuery returns the routing tables closest peers.
 func (dht *IpfsDHT) nearestPeersToQuery(pmes *pb.Message, count int) []peer.ID {
+	if pmes.GetType() == pb.Message_GET_PROVIDERS {
+		// for GET_PROVIDERS messages, the message key is the hashed multihash, so don't hash it again
+		closer := dht.routingTable.NearestPeers(kb.ID(string(pmes.GetKey())), count)
+		return closer
+	}
+
 	closer := dht.routingTable.NearestPeers(kb.ConvertKey(string(pmes.GetKey())), count)
 	return closer
 }

dht_test.go

@@ -26,6 +26,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"github.com/libp2p/go-libp2p-kad-dht/internal"
 	test "github.com/libp2p/go-libp2p-kad-dht/internal/testing"
 	pb "github.com/libp2p/go-libp2p-kad-dht/pb"
 
@@ -1298,7 +1299,11 @@ func TestClientModeConnect(t *testing.T) {
 
 	c := testCaseCids[0]
 	p := peer.ID("TestPeer")
-	a.ProviderStore().AddProvider(ctx, c.Hash(), peer.AddrInfo{ID: p})
+	mhHash := internal.Sha256Multihash(c.Hash())
+	err := a.ProviderStore().AddProvider(ctx, mhHash[:], peer.AddrInfo{ID: p})
+	if err != nil {
+		t.Fatal(err)
+	}
 	time.Sleep(time.Millisecond * 5) // just in case...
 
 	provs, err := b.FindProviders(ctx, c)

dual/dual_test.go

@@ -9,7 +9,7 @@ import (
 	u "github.com/ipfs/go-ipfs-util"
 	dht "github.com/libp2p/go-libp2p-kad-dht"
 	test "github.com/libp2p/go-libp2p-kad-dht/internal/testing"
-	record "github.com/libp2p/go-libp2p-record"
+	record "github.com/libp2p/go-libp2p-record" //nolint:typecheck
 	"github.com/libp2p/go-libp2p/core/host"
 	"github.com/libp2p/go-libp2p/core/peer"
 	peerstore "github.com/libp2p/go-libp2p/core/peerstore"

fullrt/dht.go

@@ -777,10 +777,14 @@ func (dht *FullRT) Provide(ctx context.Context, key cid.Cid, brdcst bool) (err e
 		return fmt.Errorf("invalid cid: undefined")
 	}
 	keyMH := key.Hash()
-	logger.Debugw("providing", "cid", key, "mh", internal.LoggableProviderRecordBytes(keyMH))
+	mhHash := internal.Sha256Multihash(keyMH)
+	logger.Debugw("providing", "cid", key, "mh", internal.LoggableProviderRecordBytes(keyMH), "mhHash", mhHash)
 
 	// add self locally
-	dht.ProviderManager.AddProvider(ctx, keyMH, peer.AddrInfo{ID: dht.h.ID()})
+	err = dht.ProviderManager.AddProvider(ctx, keyMH, peer.AddrInfo{ID: dht.h.ID()})
+	if err != nil {
+		return err
+	}
 	if !brdcst {
 		return nil
 	}

go.mod

@@ -3,6 +3,7 @@ module github.com/libp2p/go-libp2p-kad-dht
 go 1.18
 
 require (
+	github.com/ChainSafe/go-keyspace v0.0.0-20220830203127-eae3b7c602f0
 	github.com/gogo/protobuf v1.3.2
 	github.com/google/gopacket v1.1.19
 	github.com/google/uuid v1.3.0
@@ -16,6 +17,7 @@ require (
 	github.com/ipfs/go-log v1.0.5
 	github.com/jbenet/goprocess v0.1.4
 	github.com/libp2p/go-libp2p v0.22.0
+	github.com/libp2p/go-libp2p-core v0.20.0
 	github.com/libp2p/go-libp2p-kbucket v0.4.7
 	github.com/libp2p/go-libp2p-record v0.2.0
 	github.com/libp2p/go-libp2p-routing-helpers v0.2.3
@@ -29,7 +31,6 @@ require (
 	github.com/multiformats/go-multihash v0.2.1
 	github.com/multiformats/go-multistream v0.3.3
 	github.com/stretchr/testify v1.8.0
-	github.com/whyrusleeping/go-keyspace v0.0.0-20160322163242-5b898ac5add1
 	go.opencensus.io v0.23.0
 	go.opentelemetry.io/otel v1.7.0
 	go.opentelemetry.io/otel/trace v1.7.0
@@ -70,7 +71,6 @@ require (
 	github.com/libp2p/go-cidranger v1.1.0 // indirect
 	github.com/libp2p/go-flow-metrics v0.1.0 // indirect
 	github.com/libp2p/go-libp2p-asn-util v0.2.0 // indirect
-	github.com/libp2p/go-libp2p-core v0.20.0 // indirect
 	github.com/libp2p/go-libp2p-peerstore v0.8.0 // indirect
 	github.com/libp2p/go-nat v0.1.0 // indirect
 	github.com/libp2p/go-openssl v0.1.0 // indirect

go.sum

@@ -41,6 +41,8 @@ git.apache.org/thrift.git v0.0.0-20180902110319-2566ecd5d999/go.mod h1:fPE2ZNJGy
 github.com/AndreasBriese/bbloom v0.0.0-20190306092124-e2d15f34fcf9/go.mod h1:bOvUY6CB00SOBii9/FifXqc0awNKxLFCL/+pkDPuyl8=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo=
+github.com/ChainSafe/go-keyspace v0.0.0-20220830203127-eae3b7c602f0 h1:iTTaKZ4D0mS4VWOPJNzQXMhDVNtTAjK8GGCVIz46XtA=
+github.com/ChainSafe/go-keyspace v0.0.0-20220830203127-eae3b7c602f0/go.mod h1:NE6Gpl6geZ3MkYhDdbEVhbuNAtISJJx6MRv5l23E22Y=
 github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU=
 github.com/aead/siphash v1.0.1/go.mod h1:Nywa3cDsYNNK3gaciGTWPwHt0wlpNV15vwmswBAUSII=
 github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
@@ -643,7 +645,6 @@ github.com/viant/toolbox v0.24.0/go.mod h1:OxMCG57V0PXuIP2HNQrtJf2CjqdmbrOx5EkMI
 github.com/wangjia184/sortedset v0.0.0-20160527075905-f5d03557ba30/go.mod h1:YkocrP2K2tcw938x9gCOmT5G5eCD6jsTz0SZuyAqwIE=
 github.com/warpfork/go-wish v0.0.0-20200122115046-b9ea61034e4a h1:G++j5e0OC488te356JvdhaM8YS6nMsjLAYF7JxCv07w=
 github.com/warpfork/go-wish v0.0.0-20200122115046-b9ea61034e4a/go.mod h1:x6AKhvSSexNrVSrViXSHUEbICjmGXhtgABaHIySUSGw=
-github.com/whyrusleeping/go-keyspace v0.0.0-20160322163242-5b898ac5add1 h1:EKhdznlJHPMoKr0XTrX+IlJs1LH3lyx2nfr1dOlZ79k=
 github.com/whyrusleeping/go-keyspace v0.0.0-20160322163242-5b898ac5add1/go.mod h1:8UvriyWtv5Q5EOgjHaSseUEdkQfvwFv1I/In/O2M9gc=
 github.com/whyrusleeping/go-logging v0.0.0-20170515211332-0457bb6b88fc/go.mod h1:bopw91TMyo8J3tvftk8xmU2kPmlrt4nScJQZU2hE5EM=
 github.com/x-cray/logrus-prefixed-formatter v0.5.2/go.mod h1:2duySbKsL6M18s5GU7VPsoEPHyzalCE06qoARUCeBBE=

handlers.go

@@ -359,7 +359,10 @@ func (dht *IpfsDHT) handleAddProvider(ctx context.Context, p peer.ID, pmes *pb.M
 			continue
 		}
 
-		dht.providerStore.AddProvider(ctx, key, peer.AddrInfo{ID: p})
+		err := dht.providerStore.AddProvider(ctx, key, peer.AddrInfo{ID: p})
+		if err != nil {
+			return nil, err
+		}
 	}
 
 	return nil, nil

internal/utils.go

@@ -0,0 +1,13 @@
+package internal
+
+import (
+	"crypto/sha256"
+
+	"github.com/multiformats/go-multihash"
+)
+
+type Hash [32]byte
+
+func Sha256Multihash(mh multihash.Multihash) Hash {
+	return sha256.Sum256(mh)
+}

lookup.go

@@ -11,7 +11,7 @@ import (
 	kb "github.com/libp2p/go-libp2p-kbucket"
 )
 
-// GetClosestPeers is a Kademlia 'node lookup' operation. Returns a channel of
+// GetClosestPeers is a Kademlia 'node lookup' operation. Returns a list of
 // the K closest peers to the given key.
 //
 // If the context is canceled, this function will return the context error along
@@ -20,8 +20,11 @@ func (dht *IpfsDHT) GetClosestPeers(ctx context.Context, key string) ([]peer.ID,
 	if key == "" {
 		return nil, fmt.Errorf("can't lookup empty key")
 	}
-	// TODO: I can break the interface! return []peer.ID
-	lookupRes, err := dht.runLookupWithFollowup(ctx, key,
+
+	const isHashed = false
+
+	//TODO: I can break the interface! return []peer.ID
+	lookupRes, err := dht.runLookupWithFollowup(ctx, key, isHashed,
 		func(ctx context.Context, p peer.ID) ([]*peer.AddrInfo, error) {
 			// For DHT query command
 			routing.PublishQueryEvent(ctx, &routing.QueryEvent{

providers/providers_manager_test.go

@@ -36,7 +36,10 @@ func TestProviderManager(t *testing.T) {
 		t.Fatal(err)
 	}
 	a := u.Hash([]byte("test"))
-	p.AddProvider(ctx, a, peer.AddrInfo{ID: peer.ID("testingprovider")})
+	err = p.AddProvider(ctx, a, peer.AddrInfo{ID: peer.ID("testingprovider")})
+	if err != nil {
+		t.Fatal(err)
+	}
 
 	// Not cached
 	// TODO verify that cache is empty
@@ -52,8 +55,15 @@ func TestProviderManager(t *testing.T) {
 		t.Fatal("Could not retrieve provider.")
 	}
 
-	p.AddProvider(ctx, a, peer.AddrInfo{ID: peer.ID("testingprovider2")})
-	p.AddProvider(ctx, a, peer.AddrInfo{ID: peer.ID("testingprovider3")})
+	err = p.AddProvider(ctx, a, peer.AddrInfo{ID: peer.ID("testingprovider2")})
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = p.AddProvider(ctx, a, peer.AddrInfo{ID: peer.ID("testingprovider3")})
+	if err != nil {
+		t.Fatal(err)
+	}
+
 	// TODO verify that cache is already up to date
 	resp, _ = p.GetProviders(ctx, a)
 	if len(resp) != 3 {
@@ -88,7 +98,10 @@ func TestProvidersDatastore(t *testing.T) {
 	for i := 0; i < 100; i++ {
 		h := u.Hash([]byte(fmt.Sprint(i)))
 		mhs = append(mhs, h)
-		p.AddProvider(ctx, h, peer.AddrInfo{ID: friend})
+		err = p.AddProvider(ctx, h, peer.AddrInfo{ID: friend})
+		if err != nil {
+			t.Fatal(err)
+		}
 	}
 
 	for _, c := range mhs {
@@ -179,15 +192,27 @@ func TestProvidesExpire(t *testing.T) {
 	}
 
 	for _, h := range mhs[:5] {
-		p.AddProvider(ctx, h, peer.AddrInfo{ID: peers[0]})
-		p.AddProvider(ctx, h, peer.AddrInfo{ID: peers[1]})
+		err = p.AddProvider(ctx, h, peer.AddrInfo{ID: peers[0]})
+		if err != nil {
+			t.Fatal(err)
+		}
+		err = p.AddProvider(ctx, h, peer.AddrInfo{ID: peers[1]})
+		if err != nil {
+			t.Fatal(err)
+		}
 	}
 
 	time.Sleep(time.Second / 4)
 
 	for _, h := range mhs[5:] {
-		p.AddProvider(ctx, h, peer.AddrInfo{ID: peers[0]})
-		p.AddProvider(ctx, h, peer.AddrInfo{ID: peers[1]})
+		err = p.AddProvider(ctx, h, peer.AddrInfo{ID: peers[0]})
+		if err != nil {
+			t.Fatal(err)
+		}
+		err = p.AddProvider(ctx, h, peer.AddrInfo{ID: peers[1]})
+		if err != nil {
+			t.Fatal(err)
+		}
 	}
 
 	for _, h := range mhs {
@@ -289,7 +314,10 @@ func TestLargeProvidersSet(t *testing.T) {
 		h := u.Hash([]byte(fmt.Sprint(i)))
 		mhs = append(mhs, h)
 		for _, pid := range peers {
-			p.AddProvider(ctx, h, peer.AddrInfo{ID: pid})
+			err = p.AddProvider(ctx, h, peer.AddrInfo{ID: pid})
+			if err != nil {
+				t.Fatal(err)
+			}
 		}
 	}
 
@@ -324,11 +352,20 @@ func TestUponCacheMissProvidersAreReadFromDatastore(t *testing.T) {
 	}
 
 	// add provider
-	pm.AddProvider(ctx, h1, peer.AddrInfo{ID: p1})
+	err = pm.AddProvider(ctx, h1, peer.AddrInfo{ID: p1})
+	if err != nil {
+		t.Fatal(err)
+	}
 	// make the cached provider for h1 go to datastore
-	pm.AddProvider(ctx, h2, peer.AddrInfo{ID: p1})
+	err = pm.AddProvider(ctx, h2, peer.AddrInfo{ID: p1})
+	if err != nil {
+		t.Fatal(err)
+	}
 	// now just offloaded record should be brought back and joined with p2
-	pm.AddProvider(ctx, h1, peer.AddrInfo{ID: p2})
+	err = pm.AddProvider(ctx, h1, peer.AddrInfo{ID: p2})
+	if err != nil {
+		t.Fatal(err)
+	}
 
 	h1Provs, _ := pm.GetProviders(ctx, h1)
 	if len(h1Provs) != 2 {
@@ -353,11 +390,20 @@ func TestWriteUpdatesCache(t *testing.T) {
 	}
 
 	// add provider
-	pm.AddProvider(ctx, h1, peer.AddrInfo{ID: p1})
+	err = pm.AddProvider(ctx, h1, peer.AddrInfo{ID: p1})
+	if err != nil {
+		t.Fatal(err)
+	}
 	// force into the cache
-	pm.GetProviders(ctx, h1)
+	_, err = pm.GetProviders(ctx, h1)
+	if err != nil {
+		t.Fatal(err)
+	}
 	// add a second provider
-	pm.AddProvider(ctx, h1, peer.AddrInfo{ID: p2})
+	err = pm.AddProvider(ctx, h1, peer.AddrInfo{ID: p2})
+	if err != nil {
+		t.Fatal(err)
+	}
 
 	c1Provs, _ := pm.GetProviders(ctx, h1)
 	if len(c1Provs) != 2 {

qpeerset/qpeerset.go

@@ -4,8 +4,8 @@ import (
 	"math/big"
 	"sort"
 
-	"github.com/libp2p/go-libp2p/core/peer"
-	ks "github.com/whyrusleeping/go-keyspace"
+	ks "github.com/ChainSafe/go-keyspace"
+	"github.com/libp2p/go-libp2p-core/peer"
 )
 
 // PeerState describes the state of a peer ID during the lifecycle of an individual lookup.
@@ -67,6 +67,14 @@ func NewQueryPeerset(key string) *QueryPeerset {
 	}
 }
 
+func NewQueryPeersetFromHash(hash [32]byte) *QueryPeerset {
+	return &QueryPeerset{
+		key:    ks.XORKeySpace.KeyFromHash(hash),
+		all:    []queryPeerState{},
+		sorted: false,
+	}
+}
+
 func (qp *QueryPeerset) find(p peer.ID) int {
 	for i := range qp.all {
 		if qp.all[i].id == p {

query.go

@@ -76,9 +76,9 @@ type lookupWithFollowupResult struct {
 //
 // After the lookup is complete the query function is run (unless stopped) against all of the top K peers from the
 // lookup that have not already been successfully queried.
-func (dht *IpfsDHT) runLookupWithFollowup(ctx context.Context, target string, queryFn queryFn, stopFn stopFn) (*lookupWithFollowupResult, error) {
+func (dht *IpfsDHT) runLookupWithFollowup(ctx context.Context, target string, isHashed bool, queryFn queryFn, stopFn stopFn) (*lookupWithFollowupResult, error) {
 	// run the query
-	lookupRes, err := dht.runQuery(ctx, target, queryFn, stopFn)
+	lookupRes, err := dht.runQuery(ctx, target, isHashed, queryFn, stopFn)
 	if err != nil {
 		return nil, err
 	}
@@ -145,9 +145,15 @@ processFollowUp:
 	return lookupRes, nil
 }
 
-func (dht *IpfsDHT) runQuery(ctx context.Context, target string, queryFn queryFn, stopFn stopFn) (*lookupWithFollowupResult, error) {
+func (dht *IpfsDHT) runQuery(ctx context.Context, target string, isHashed bool, queryFn queryFn, stopFn stopFn) (*lookupWithFollowupResult, error) {
 	// pick the K closest peers to the key in our Routing table.
-	targetKadID := kb.ConvertKey(target)
+
+	targetKadID := kb.ID(target)
+	if !isHashed {
+		// this sha256 hashes `target`, so don't hash `target` if it's already been hashed
+		targetKadID = kb.ConvertKey(target)
+	}
+
 	seedPeers := dht.routingTable.NearestPeers(targetKadID, dht.bucketSize)
 	if len(seedPeers) == 0 {
 		routing.PublishQueryEvent(ctx, &routing.QueryEvent{
@@ -157,12 +163,21 @@ func (dht *IpfsDHT) runQuery(ctx context.Context, target string, queryFn queryFn
 		return nil, kb.ErrLookupFailure
 	}
 
+	var qpset *qpeerset.QueryPeerset
+	if isHashed {
+		var hash [32]byte
+		copy(hash[:], targetKadID)
+		qpset = qpeerset.NewQueryPeersetFromHash(hash)
+	} else {
+		qpset = qpeerset.NewQueryPeerset(target)
+	}
+
 	q := &query{
 		id:         uuid.New(),
 		key:        target,
 		ctx:        ctx,
 		dht:        dht,
-		queryPeers: qpeerset.NewQueryPeerset(target),
+		queryPeers: qpset,
 		seedPeers:  seedPeers,
 		peerTimes:  make(map[peer.ID]time.Duration),
 		terminated: false,