feat: go-libp2p 0.16 and autosharding and go-datastore to take contexts

.circleci/main.yml

@@ -56,7 +56,7 @@ executors:
     environment:
       <<: *default_environment
       NO_SANDBOX: true
-      IPFS_REUSEPORT: false
+      LIBP2P_TCP_REUSEPORT: false
       LIBP2P_ALLOW_WEAK_RSA_KEYS: 1
       E2E_IPFSD_TYPE: go
   dockerizer:
@@ -216,7 +216,8 @@ jobs:
             - bin/ipfs
       - *store_gomod
   interop:
-    executor: node
+    docker:
+      - image: cimg/go:1.16-node
     parallelism: 4
     steps:
       - *make_out_dirs
@@ -227,7 +228,7 @@ jobs:
           command: |
             npm init -y
             npm install ipfs@^0.59.1
-            npm install ipfs-interop@^7.0.3
+            npm install ipfs/interop#feat/go-libp2p-0.16-and-sharding-config-option
             npm install [email protected]
           working_directory: ~/ipfs/go-ipfs/interop
       - run:
@@ -238,7 +239,7 @@ jobs:
             npx ipfs-interop -- -t node -f $(sed -n -e "s|^require('\(.*\)')$|test/\1|p" node_modules/ipfs-interop/test/node.js | circleci tests split) -- --reporter mocha-circleci-reporter
           working_directory: ~/ipfs/go-ipfs/interop
           environment:
-            IPFS_REUSEPORT: false
+            LIBP2P_TCP_REUSEPORT: false
             LIBP2P_ALLOW_WEAK_RSA_KEYS: 1
             IPFS_GO_EXEC: /tmp/circleci-workspace/bin/ipfs
       - store_test_results:

Rules.mk

@@ -22,7 +22,7 @@ include mk/golang.mk
 ifeq ($(TEST_NO_FUSE),1)
 	GOTAGS += nofuse
 endif
-export IPFS_REUSEPORT=false
+export LIBP2P_TCP_REUSEPORT=false
 
 # -------------------- #
 #       sub-files      #

blocks/blockstoreutil/remove.go

@@ -8,7 +8,7 @@ import (
 
 	cid "github.com/ipfs/go-cid"
 	bs "github.com/ipfs/go-ipfs-blockstore"
-	"github.com/ipfs/go-ipfs-pinner"
+	pin "github.com/ipfs/go-ipfs-pinner"
 )
 
 // RemovedBlock is used to represent the result of removing a block.
@@ -40,15 +40,15 @@ func RmBlocks(ctx context.Context, blocks bs.GCBlockstore, pins pin.Pinner, cids
 	go func() {
 		defer close(out)
 
-		unlocker := blocks.GCLock()
-		defer unlocker.Unlock()
+		unlocker := blocks.GCLock(ctx)
+		defer unlocker.Unlock(ctx)
 
 		stillOkay := FilterPinned(ctx, pins, out, cids)
 
 		for _, c := range stillOkay {
 			// Kept for backwards compatibility. We may want to
 			// remove this sometime in the future.
-			has, err := blocks.Has(c)
+			has, err := blocks.Has(ctx, c)
 			if err != nil {
 				out <- &RemovedBlock{Hash: c.String(), Error: err.Error()}
 				continue
@@ -58,7 +58,7 @@ func RmBlocks(ctx context.Context, blocks bs.GCBlockstore, pins pin.Pinner, cids
 				continue
 			}
 
-			err = blocks.DeleteBlock(c)
+			err = blocks.DeleteBlock(ctx, c)
 			if err != nil {
 				out <- &RemovedBlock{Hash: c.String(), Error: err.Error()}
 			} else if !opts.Quiet {

core/commands/dag/import.go

@@ -42,8 +42,8 @@ func dagImport(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment
 	// This is especially important for use cases like dagger:
 	//    ipfs dag import $( ... | ipfs-dagger --stdout=carfifos )
 	//
-	unlocker := node.Blockstore.PinLock()
-	defer unlocker.Unlock()
+	unlocker := node.Blockstore.PinLock(req.Context)
+	defer unlocker.Unlock(req.Context)
 
 	doPinRoots, _ := req.Options[pinRootsOptionName].(bool)
 
@@ -87,7 +87,7 @@ func dagImport(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment
 
 			ret := RootMeta{Cid: c}
 
-			if block, err := node.Blockstore.Get(c); err != nil {
+			if block, err := node.Blockstore.Get(req.Context, c); err != nil {
 				ret.PinErrorMsg = err.Error()
 			} else if nd, err := ipld.Decode(block); err != nil {
 				ret.PinErrorMsg = err.Error()

core/commands/dht.go

@@ -273,7 +273,7 @@ var provideRefDhtCmd = &cmds.Command{
 				return err
 			}
 
-			has, err := nd.Blockstore.Has(c)
+			has, err := nd.Blockstore.Has(req.Context, c)
 			if err != nil {
 				return err
 			}

core/commands/filestore.go

@@ -1,17 +1,18 @@
 package commands
 
 import (
+	"context"
 	"fmt"
 	"io"
 	"os"
 
 	filestore "github.com/ipfs/go-filestore"
+	cmds "github.com/ipfs/go-ipfs-cmds"
 	core "github.com/ipfs/go-ipfs/core"
 	cmdenv "github.com/ipfs/go-ipfs/core/commands/cmdenv"
 	e "github.com/ipfs/go-ipfs/core/commands/e"
 
 	"github.com/ipfs/go-cid"
-	"github.com/ipfs/go-ipfs-cmds"
 )
 
 var FileStoreCmd = &cmds.Command{
@@ -56,17 +57,17 @@ The output is:
 		}
 		args := req.Arguments
 		if len(args) > 0 {
-			return listByArgs(res, fs, args)
+			return listByArgs(req.Context, res, fs, args)
 		}
 
 		fileOrder, _ := req.Options[fileOrderOptionName].(bool)
-		next, err := filestore.ListAll(fs, fileOrder)
+		next, err := filestore.ListAll(req.Context, fs, fileOrder)
 		if err != nil {
 			return err
 		}
 
 		for {
-			r := next()
+			r := next(req.Context)
 			if r == nil {
 				break
 			}
@@ -133,17 +134,17 @@ For ERROR entries the error will also be printed to stderr.
 		}
 		args := req.Arguments
 		if len(args) > 0 {
-			return listByArgs(res, fs, args)
+			return listByArgs(req.Context, res, fs, args)
 		}
 
 		fileOrder, _ := req.Options[fileOrderOptionName].(bool)
-		next, err := filestore.VerifyAll(fs, fileOrder)
+		next, err := filestore.VerifyAll(req.Context, fs, fileOrder)
 		if err != nil {
 			return err
 		}
 
 		for {
-			r := next()
+			r := next(req.Context)
 			if r == nil {
 				break
 			}
@@ -206,7 +207,7 @@ var dupsFileStore = &cmds.Command{
 		}
 
 		for cid := range ch {
-			have, err := fs.MainBlockstore().Has(cid)
+			have, err := fs.MainBlockstore().Has(req.Context, cid)
 			if err != nil {
 				return res.Emit(&RefWrapper{Err: err.Error()})
 			}
@@ -235,7 +236,7 @@ func getFilestore(env cmds.Environment) (*core.IpfsNode, *filestore.Filestore, e
 	return n, fs, err
 }
 
-func listByArgs(res cmds.ResponseEmitter, fs *filestore.Filestore, args []string) error {
+func listByArgs(ctx context.Context, res cmds.ResponseEmitter, fs *filestore.Filestore, args []string) error {
 	for _, arg := range args {
 		c, err := cid.Decode(arg)
 		if err != nil {
@@ -248,7 +249,7 @@ func listByArgs(res cmds.ResponseEmitter, fs *filestore.Filestore, args []string
 			}
 			continue
 		}
-		r := filestore.Verify(fs, c)
+		r := filestore.Verify(ctx, fs, c)
 		if err := res.Emit(r); err != nil {
 			return err
 		}

core/commands/pin/remotepin.go

@@ -174,7 +174,7 @@ NOTE: a comma-separated notation is supported in CLI for convenience:
 			return err
 		}
 
-		isInBlockstore, err := node.Blockstore.Has(rp.Cid())
+		isInBlockstore, err := node.Blockstore.Has(req.Context, rp.Cid())
 		if err != nil {
 			return err
 		}

core/commands/repo.go

@@ -239,7 +239,7 @@ func verifyWorkerRun(ctx context.Context, wg *sync.WaitGroup, keys <-chan cid.Ci
 	defer wg.Done()
 
 	for k := range keys {
-		_, err := bs.Get(k)
+		_, err := bs.Get(ctx, k)
 		if err != nil {
 			select {
 			case results <- fmt.Sprintf("block %s was corrupt (%s)", k, err):

core/coreapi/block.go

@@ -46,10 +46,10 @@ func (api *BlockAPI) Put(ctx context.Context, src io.Reader, opts ...caopts.Bloc
 	}
 
 	if settings.Pin {
-		defer api.blockstore.PinLock().Unlock()
+		defer api.blockstore.PinLock(ctx).Unlock(ctx)
 	}
 
-	err = api.blocks.AddBlock(b)
+	err = api.blocks.AddBlock(ctx, b)
 	if err != nil {
 		return nil, err
 	}

core/coreapi/dag.go

@@ -4,7 +4,7 @@ import (
 	"context"
 
 	cid "github.com/ipfs/go-cid"
-	"github.com/ipfs/go-ipfs-pinner"
+	pin "github.com/ipfs/go-ipfs-pinner"
 	ipld "github.com/ipfs/go-ipld-format"
 	dag "github.com/ipfs/go-merkledag"
 )
@@ -18,7 +18,7 @@ type dagAPI struct {
 type pinningAdder CoreAPI
 
 func (adder *pinningAdder) Add(ctx context.Context, nd ipld.Node) error {
-	defer adder.blockstore.PinLock().Unlock()
+	defer adder.blockstore.PinLock(ctx).Unlock(ctx)
 
 	if err := adder.dag.Add(ctx, nd); err != nil {
 		return err
@@ -30,7 +30,7 @@ func (adder *pinningAdder) Add(ctx context.Context, nd ipld.Node) error {
 }
 
 func (adder *pinningAdder) AddMany(ctx context.Context, nds []ipld.Node) error {
-	defer adder.blockstore.PinLock().Unlock()
+	defer adder.blockstore.PinLock(ctx).Unlock(ctx)
 
 	if err := adder.dag.AddMany(ctx, nds); err != nil {
 		return err

core/coreapi/dht.go

@@ -76,7 +76,7 @@ func (api *DhtAPI) Provide(ctx context.Context, path path.Path, opts ...caopts.D
 
 	c := rp.Cid()
 
-	has, err := api.blockstore.Has(c)
+	has, err := api.blockstore.Has(ctx, c)
 	if err != nil {
 		return err
 	}

core/coreapi/object.go

@@ -12,7 +12,7 @@ import (
 	"io/ioutil"
 
 	cid "github.com/ipfs/go-cid"
-	"github.com/ipfs/go-ipfs-pinner"
+	pin "github.com/ipfs/go-ipfs-pinner"
 	ipld "github.com/ipfs/go-ipld-format"
 	dag "github.com/ipfs/go-merkledag"
 	"github.com/ipfs/go-merkledag/dagutils"
@@ -110,7 +110,7 @@ func (api *ObjectAPI) Put(ctx context.Context, src io.Reader, opts ...caopts.Obj
 	}
 
 	if options.Pin {
-		defer api.blockstore.PinLock().Unlock()
+		defer api.blockstore.PinLock(ctx).Unlock(ctx)
 	}
 
 	err = api.dag.Add(ctx, dagnode)

core/coreapi/pin.go

@@ -27,7 +27,7 @@ func (api *PinAPI) Add(ctx context.Context, p path.Path, opts ...caopts.PinAddOp
 		return err
 	}
 
-	defer api.blockstore.PinLock().Unlock()
+	defer api.blockstore.PinLock(ctx).Unlock(ctx)
 
 	err = api.pinning.Pin(ctx, dagNode, settings.Recursive)
 	if err != nil {
@@ -89,7 +89,7 @@ func (api *PinAPI) Rm(ctx context.Context, p path.Path, opts ...caopts.PinRmOpti
 
 	// Note: after unpin the pin sets are flushed to the blockstore, so we need
 	// to take a lock to prevent a concurrent garbage collection
-	defer api.blockstore.PinLock().Unlock()
+	defer api.blockstore.PinLock(ctx).Unlock(ctx)
 
 	if err = api.pinning.Unpin(ctx, rp.Cid(), settings.Recursive); err != nil {
 		return err
@@ -114,7 +114,7 @@ func (api *PinAPI) Update(ctx context.Context, from path.Path, to path.Path, opt
 		return err
 	}
 
-	defer api.blockstore.PinLock().Unlock()
+	defer api.blockstore.PinLock(ctx).Unlock(ctx)
 
 	err = api.pinning.Update(ctx, fp.Cid(), tp.Cid(), settings.Unpin)
 	if err != nil {

core/coreapi/test/path_test.go

@@ -14,7 +14,6 @@ import (
 	"github.com/ipld/go-ipld-prime"
 )
 
-
 func TestPathUnixFSHAMTPartial(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -27,16 +26,16 @@ func TestPathUnixFSHAMTPartial(t *testing.T) {
 	a := apis[0]
 
 	// Setting this after instantiating the swarm so that it's not clobbered by loading the go-ipfs config
-	prevVal := uio.UseHAMTSharding
-	uio.UseHAMTSharding = true
+	prevVal := uio.HAMTShardingSize
+	uio.HAMTShardingSize = 1
 	defer func() {
-		uio.UseHAMTSharding = prevVal
+		uio.HAMTShardingSize = prevVal
 	}()
 
 	// Create and add a sharded directory
 	dir := make(map[string]files.Node)
 	// Make sure we have at least two levels of sharding
-	for i := 0; i < uio.DefaultShardWidth + 1; i++ {
+	for i := 0; i < uio.DefaultShardWidth+1; i++ {
 		dir[strconv.Itoa(i)] = files.NewBytesFile([]byte(strconv.Itoa(i)))
 	}
 
@@ -67,7 +66,7 @@ func TestPathUnixFSHAMTPartial(t *testing.T) {
 	for k := range dir {
 		// The node will go out to the (non-existent) network looking for the missing block. Make sure we're erroring
 		// because we exceeded the timeout on our query
-		timeoutCtx, timeoutCancel := context.WithTimeout(ctx, time.Second * 1)
+		timeoutCtx, timeoutCancel := context.WithTimeout(ctx, time.Second*1)
 		_, err := a.ResolveNode(timeoutCtx, path.Join(r, k))
 		if err != nil {
 			if timeoutCtx.Err() == nil {

core/coreapi/unixfs.go

@@ -111,10 +111,10 @@ func (api *UnixfsAPI) Add(ctx context.Context, files files.Node, opts ...options
 			DAGService: dserv,
 			syncFn: func() error {
 				ds := api.repo.Datastore()
-				if err := ds.Sync(bstore.BlockPrefix); err != nil {
+				if err := ds.Sync(ctx, bstore.BlockPrefix); err != nil {
 					return err
 				}
-				return ds.Sync(filestore.FilestorePrefix)
+				return ds.Sync(ctx, filestore.FilestorePrefix)
 			},
 		}
 	}
@@ -164,7 +164,7 @@ func (api *UnixfsAPI) Add(ctx context.Context, files files.Node, opts ...options
 		fileAdder.SetMfsRoot(mr)
 	}
 
-	nd, err := fileAdder.AddAllAndPin(files)
+	nd, err := fileAdder.AddAllAndPin(ctx, files)
 	if err != nil {
 		return nil, err
 	}

core/corehttp/metrics_test.go

@@ -21,7 +21,7 @@ func TestPeersTotal(t *testing.T) {
 	hosts := make([]*bhost.BasicHost, 4)
 	for i := 0; i < 4; i++ {
 		var err error
-		hosts[i], err = bhost.NewHost(ctx, swarmt.GenSwarm(t, ctx), nil)
+		hosts[i], err = bhost.NewHost(swarmt.GenSwarm(t), nil)
 		if err != nil {
 			t.Fatal(err)
 		}

core/corerepo/gc.go

@@ -205,7 +205,7 @@ func ConditionalGC(ctx context.Context, node *core.IpfsNode, offset uint64) erro
 }
 
 func (gc *GC) maybeGC(ctx context.Context, offset uint64) error {
-	storage, err := gc.Repo.GetStorageUsage()
+	storage, err := gc.Repo.GetStorageUsage(ctx)
 	if err != nil {
 		return err
 	}

core/corerepo/stat.go

@@ -71,7 +71,7 @@ func RepoSize(ctx context.Context, n *core.IpfsNode) (SizeStat, error) {
 		return SizeStat{}, err
 	}
 
-	usage, err := r.GetStorageUsage()
+	usage, err := r.GetStorageUsage(ctx)
 	if err != nil {
 		return SizeStat{}, err
 	}