Put peer ID of provider peer in the Indexer core metadata (#110)

* storethehash-needs-a-peer-id

* better documentation

dagstore.go

@@ -7,6 +7,8 @@ import (
 	"os"
 	"sync"
 
+	"github.com/libp2p/go-libp2p-core/peer"
+
 	mh "github.com/multiformats/go-multihash"
 
 	carindex "github.com/ipld/go-car/v2/index"
@@ -207,7 +209,7 @@ func NewDAGStore(cfg Config) (*DAGStore, error) {
 
 	if cfg.TopLevelIndex == nil {
 		log.Info("using in-memory inverted index")
-		cfg.TopLevelIndex = index.NewInverted(memory.New())
+		cfg.TopLevelIndex = index.NewInverted(memory.New(), peer.ID(""))
 	}
 
 	// handle the datastore.

go.mod

@@ -3,12 +3,13 @@ module github.com/filecoin-project/dagstore
 go 1.16
 
 require (
-	github.com/filecoin-project/go-indexer-core v0.2.3
+	github.com/filecoin-project/go-indexer-core v0.2.4
 	github.com/ipfs/go-block-format v0.0.3
 	github.com/ipfs/go-cid v0.1.0
 	github.com/ipfs/go-datastore v0.4.5
 	github.com/ipfs/go-log/v2 v2.3.0
 	github.com/ipld/go-car/v2 v2.1.0
+	github.com/libp2p/go-libp2p-core v0.9.0
 	github.com/mr-tron/base58 v1.2.0
 	github.com/multiformats/go-multicodec v0.3.1-0.20210902112759-1539a079fd61
 	github.com/multiformats/go-multihash v0.0.16

go.sum

@@ -82,11 +82,12 @@ github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.m
 github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98=
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
 github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
-github.com/filecoin-project/go-indexer-core v0.2.3 h1:kaUL2r8CuihK53lhmtCScffb7Bzs+N1yRGpwvxzCN+U=
-github.com/filecoin-project/go-indexer-core v0.2.3/go.mod h1:MSe5aRWmfRB+5syR4yDV+OApgJU+MUJ4rl9VUuzwCfc=
+github.com/filecoin-project/go-indexer-core v0.2.4 h1:90vvxoBeNZN+h4W+vZ+VsoxKaDBr/bfZJJNByapGeM0=
+github.com/filecoin-project/go-indexer-core v0.2.4/go.mod h1:MSe5aRWmfRB+5syR4yDV+OApgJU+MUJ4rl9VUuzwCfc=
 github.com/franela/goblin v0.0.0-20200105215937-c9ffbefa60db/go.mod h1:7dvUGVsVBjqR7JHJk0brhHOZYGmfBYOrK0ZhYMEtBr4=
 github.com/franela/goreq v0.0.0-20171204163338-bcd34c9993f8/go.mod h1:ZhphrRTfi2rbfLwlschooIH4+wKKDR4Pdxhh+TRoA20=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
+github.com/gammazero/keymutex v0.0.2 h1:cmpLBJHdEwn+WlR5Z/o9/BN92znSZTp5AKPQDpu1QcI=
 github.com/gammazero/keymutex v0.0.2/go.mod h1:qtzWCCLMisQUmVa4dvqHVgwfh4BP2YB7JxNDGXnsKrs=
 github.com/gammazero/radixtree v0.2.5 h1:muPQ4eEgCkUymFWPiVQRuXOQv4IhWg8YXH2r71MoqPM=
 github.com/gammazero/radixtree v0.2.5/go.mod h1:VPqqCDZ3YZZxAzUUsIF/ytFBigVWV7JIV1Stld8hri0=
@@ -270,6 +271,7 @@ github.com/ipld/go-car/v2 v2.1.0 h1:t8R/WXUSkfu1K1gpPk76mytCxsEdMjGcMIgpOq3/Cnw=
 github.com/ipld/go-car/v2 v2.1.0/go.mod h1:Xr6GwkDhv8dtOtgHzOynAkIOg0t0YiPc5DxBPppWqZA=
 github.com/ipld/go-ipld-prime v0.0.2-0.20191108012745-28a82f04c785/go.mod h1:bDDSvVz7vaK12FNvMeRYnpRFkSUPNQOiCYQezMD/P3w=
 github.com/ipld/go-ipld-prime-proto v0.0.0-20191113031812-e32bd156a1e5/go.mod h1:gcvzoEDBjwycpXt3LBE061wT9f46szXGHAmj9uoP6fU=
+github.com/ipld/go-storethehash v0.0.0-20210915160027-d72ca9b0968c h1:izfvqCuEqk2V7BRkh7GCm7lyKC2ItyAbzUu4WgNmggc=
 github.com/ipld/go-storethehash v0.0.0-20210915160027-d72ca9b0968c/go.mod h1:PwE6iq8TiWJRI3zMGA1RtkFAnrDMK93dLA5SUeu0lH8=
 github.com/jackpal/gateway v1.0.5 h1:qzXWUJfuMdlLMtt0a3Dgt+xkWQiA5itDEITVJtuSwMc=
 github.com/jackpal/gateway v1.0.5/go.mod h1:lTpwd4ACLXmpyiCTRtfiNyVnUmqT9RivzCDQetPfnjA=

index/inverted_index_impl.go

@@ -4,6 +4,8 @@ import (
 	"errors"
 	"fmt"
 
+	"github.com/libp2p/go-libp2p-core/peer"
+
 	"github.com/filecoin-project/go-indexer-core"
 
 	"github.com/multiformats/go-multihash"
@@ -16,27 +18,33 @@ var InvertedIndexErrNotFound = errors.New("multihash not found in Index")
 var _ Inverted = (*indexerCoreIndex)(nil)
 
 type indexerCoreIndex struct {
-	is indexer.Interface
+	is         indexer.Interface
+	selfPeerID peer.ID
 }
 
-func NewInverted(is indexer.Interface) *indexerCoreIndex {
+// NewInverted returns a new inverted index that uses `go-indexer-core`
+// as it's storage backend. We use `go-indexer-core` as the backend here
+// as it's been optimized to store (multihash -> Value) kind of data and
+// supports bulk updates via context ID and metadata-deduplication which are useful properties for our use case here.
+func NewInverted(is indexer.Interface, selfPeerID peer.ID) *indexerCoreIndex {
 	return &indexerCoreIndex{
-		is: is,
+		is:         is,
+		selfPeerID: selfPeerID,
 	}
 }
 
 func (d *indexerCoreIndex) AddMultihashesForShard(mhIter MultihashIterator, s shard.Key) error {
 	return mhIter.ForEach(func(mh multihash.Multihash) error {
 		// go-indexer-core appends values to the existing values we already have for the key
 		// it also takes care of de-duplicating values.
-		return d.is.Put(valueForShardKey(s), mh)
+		return d.is.Put(valueForShardKey(s, d.selfPeerID), mh)
 	})
 }
 
 func (d *indexerCoreIndex) DeleteMultihashesForShard(sk shard.Key, mhIter MultihashIterator) error {
 	return mhIter.ForEach(func(mh multihash.Multihash) error {
 		// remove the given value i.e. shard key from the index for the given multihash.
-		return d.is.Remove(valueForShardKey(sk), mh)
+		return d.is.Remove(valueForShardKey(sk, d.selfPeerID), mh)
 	})
 }
 
@@ -66,9 +74,10 @@ func shardKeyFromValue(val indexer.Value) shard.Key {
 	return shard.KeyFromString(str)
 }
 
-func valueForShardKey(key shard.Key) indexer.Value {
+func valueForShardKey(key shard.Key, selfPeerID peer.ID) indexer.Value {
 	return indexer.Value{
+		ProviderID:    selfPeerID,
 		ContextID:     []byte(key.String()),
-		MetadataBytes: []byte(key.String()),
+		MetadataBytes: []byte("N/A"),
 	}
 }

index/inverted_index_impl_test.go

@@ -1,9 +1,13 @@
 package index
 
 import (
+	"crypto/rand"
 	"testing"
 
-	"github.com/filecoin-project/go-indexer-core/store/memory"
+	"github.com/libp2p/go-libp2p-core/crypto"
+	"github.com/libp2p/go-libp2p-core/peer"
+
+	"github.com/filecoin-project/go-indexer-core/store/storethehash"
 
 	"golang.org/x/xerrors"
 
@@ -20,11 +24,9 @@ func TestDatastoreIndexEmpty(t *testing.T) {
 	cid1, err := cid.Parse("Qmard76Snyj9VCJBzLSLYzXnJJ2BnyCN2KAfAkpLXyt1q7")
 	req.NoError(err)
 
-	idx := NewInverted(memory.New())
-
-	l, err := idx.Size()
-	req.NoError(err)
-	req.EqualValues(0, l)
+	store, err := storethehash.New(t.TempDir())
+	require.NoError(t, err)
+	idx := NewInverted(store, genRandPeer(t))
 
 	_, err = idx.GetShardsForMultihash(cid1.Hash())
 	req.True(xerrors.Is(err, InvertedIndexErrNotFound))
@@ -44,7 +46,9 @@ func TestDatastoreIndex(t *testing.T) {
 	h2 := cid2.Hash()
 	h3 := cid3.Hash()
 
-	idx := NewInverted(memory.New())
+	store, err := storethehash.New(t.TempDir())
+	require.NoError(t, err)
+	idx := NewInverted(store, genRandPeer(t))
 
 	// Add hash to shard key mappings for h1, h2:
 	// h1 -> [shard-key-1]
@@ -89,7 +93,9 @@ func TestDatastoreIndexDelete(t *testing.T) {
 	h1 := cid1.Hash()
 	h2 := cid2.Hash()
 
-	idx := NewInverted(memory.New())
+	store, err := storethehash.New(t.TempDir())
+	require.NoError(t, err)
+	idx := NewInverted(store, genRandPeer(t))
 
 	// Add hash to shard key mappings for h1, h2:
 	// h1 -> [shard-key-1]
@@ -124,3 +130,12 @@ func (mi *mhIt) ForEach(f func(mh multihash.Multihash) error) error {
 	}
 	return nil
 }
+
+func genRandPeer(t *testing.T) peer.ID {
+	priv, _, err := crypto.GenerateRSAKeyPair(2048, rand.Reader)
+	require.NoError(t, err)
+
+	id, err := peer.IDFromPrivateKey(priv)
+	require.NoError(t, err)
+	return id
+}