Merge pull request #5282 from filecoin-project/chore/snake_context_th…

…rough_blockstore_init Snake a context through the Chain-blockstore creation
filecoin-project · Jan 30, 2021 · a541a25 · a541a25
2 parents 7c7301f + af429db
commit a541a25
Show file tree

Hide file tree

Showing 24 changed files with 71 additions and 47 deletions.
diff --git a/chain/gen/gen.go b/chain/gen/gen.go
@@ -120,12 +120,12 @@ func NewGeneratorWithSectors(numSectors int) (*ChainGen, error) {
 		return nil, xerrors.Errorf("taking mem-repo lock failed: %w", err)
 	}
 
-	ds, err := lr.Datastore("/metadata")
+	ds, err := lr.Datastore(context.TODO(), "/metadata")
 	if err != nil {
 		return nil, xerrors.Errorf("failed to get metadata datastore: %w", err)
 	}
 
-	bs, err := lr.Blockstore(repo.BlockstoreChain)
+	bs, err := lr.Blockstore(context.TODO(), repo.BlockstoreChain)
 	if err != nil {
 		return nil, err
 	}

diff --git a/chain/store/store_test.go b/chain/store/store_test.go
@@ -52,7 +52,7 @@ func BenchmarkGetRandomness(b *testing.B) {
 		b.Fatal(err)
 	}
 
-	bs, err := lr.Blockstore(repo.BlockstoreChain)
+	bs, err := lr.Blockstore(context.TODO(), repo.BlockstoreChain)
 	if err != nil {
 		b.Fatal(err)
 	}
@@ -65,7 +65,7 @@ func BenchmarkGetRandomness(b *testing.B) {
 		}
 	}()
 
-	mds, err := lr.Datastore("/metadata")
+	mds, err := lr.Datastore(context.Background(), "/metadata")
 	if err != nil {
 		b.Fatal(err)
 	}

diff --git a/cli/backup.go b/cli/backup.go
@@ -46,7 +46,7 @@ func BackupCmd(repoFlag string, rt repo.RepoType, getApi BackupApiFn) *cli.Comma
 		}
 		defer lr.Close() // nolint:errcheck
 
-		mds, err := lr.Datastore("/metadata")
+		mds, err := lr.Datastore(context.TODO(), "/metadata")
 		if err != nil {
 			return xerrors.Errorf("getting metadata datastore: %w", err)
 		}

diff --git a/cmd/lotus-seal-worker/main.go b/cmd/lotus-seal-worker/main.go
@@ -308,7 +308,7 @@ var runCmd = &cli.Command{
 
 			{
 				// init datastore for r.Exists
-				_, err := lr.Datastore("/metadata")
+				_, err := lr.Datastore(context.Background(), "/metadata")
 				if err != nil {
 					return err
 				}
@@ -327,7 +327,7 @@ var runCmd = &cli.Command{
 				log.Error("closing repo", err)
 			}
 		}()
-		ds, err := lr.Datastore("/metadata")
+		ds, err := lr.Datastore(context.Background(), "/metadata")
 		if err != nil {
 			return err
 		}

diff --git a/cmd/lotus-shed/balances.go b/cmd/lotus-shed/balances.go
@@ -175,7 +175,7 @@ var chainBalanceStateCmd = &cli.Command{
 
 		defer lkrepo.Close() //nolint:errcheck
 
-		bs, err := lkrepo.Blockstore(repo.BlockstoreChain)
+		bs, err := lkrepo.Blockstore(ctx, repo.BlockstoreChain)
 		if err != nil {
 			return fmt.Errorf("failed to open blockstore: %w", err)
 		}
@@ -188,7 +188,7 @@ var chainBalanceStateCmd = &cli.Command{
 			}
 		}()
 
-		mds, err := lkrepo.Datastore("/metadata")
+		mds, err := lkrepo.Datastore(context.Background(), "/metadata")
 		if err != nil {
 			return err
 		}
@@ -396,7 +396,7 @@ var chainPledgeCmd = &cli.Command{
 
 		defer lkrepo.Close() //nolint:errcheck
 
-		bs, err := lkrepo.Blockstore(repo.BlockstoreChain)
+		bs, err := lkrepo.Blockstore(ctx, repo.BlockstoreChain)
 		if err != nil {
 			return xerrors.Errorf("failed to open blockstore: %w", err)
 		}
@@ -409,7 +409,7 @@ var chainPledgeCmd = &cli.Command{
 			}
 		}()
 
-		mds, err := lkrepo.Datastore("/metadata")
+		mds, err := lkrepo.Datastore(context.Background(), "/metadata")
 		if err != nil {
 			return err
 		}

diff --git a/cmd/lotus-shed/datastore.go b/cmd/lotus-shed/datastore.go
@@ -2,6 +2,7 @@ package main
 
 import (
 	"bufio"
+	"context"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -75,7 +76,7 @@ var datastoreListCmd = &cli.Command{
 		}
 		defer lr.Close() //nolint:errcheck
 
-		ds, err := lr.Datastore(datastore.NewKey(cctx.Args().First()).String())
+		ds, err := lr.Datastore(context.Background(), datastore.NewKey(cctx.Args().First()).String())
 		if err != nil {
 			return err
 		}
@@ -141,7 +142,7 @@ var datastoreGetCmd = &cli.Command{
 		}
 		defer lr.Close() //nolint:errcheck
 
-		ds, err := lr.Datastore(datastore.NewKey(cctx.Args().First()).String())
+		ds, err := lr.Datastore(context.Background(), datastore.NewKey(cctx.Args().First()).String())
 		if err != nil {
 			return err
 		}

diff --git a/cmd/lotus-shed/export.go b/cmd/lotus-shed/export.go
@@ -72,7 +72,7 @@ var exportChainCmd = &cli.Command{
 
 		defer fi.Close() //nolint:errcheck
 
-		bs, err := lr.Blockstore(repo.BlockstoreChain)
+		bs, err := lr.Blockstore(ctx, repo.BlockstoreChain)
 		if err != nil {
 			return fmt.Errorf("failed to open blockstore: %w", err)
 		}
@@ -85,7 +85,7 @@ var exportChainCmd = &cli.Command{
 			}
 		}()
 
-		mds, err := lr.Datastore("/metadata")
+		mds, err := lr.Datastore(context.Background(), "/metadata")
 		if err != nil {
 			return err
 		}

diff --git a/cmd/lotus-shed/import-car.go b/cmd/lotus-shed/import-car.go
@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	"encoding/hex"
 	"fmt"
 	"io"
@@ -24,6 +25,8 @@ var importCarCmd = &cli.Command{
 			return xerrors.Errorf("opening fs repo: %w", err)
 		}
 
+		ctx := context.TODO()
+
 		exists, err := r.Exists()
 		if err != nil {
 			return err
@@ -44,7 +47,7 @@ var importCarCmd = &cli.Command{
 			return xerrors.Errorf("opening the car file: %w", err)
 		}
 
-		bs, err := lr.Blockstore(repo.BlockstoreChain)
+		bs, err := lr.Blockstore(ctx, repo.BlockstoreChain)
 		if err != nil {
 			return err
 		}
@@ -99,6 +102,8 @@ var importObjectCmd = &cli.Command{
 			return xerrors.Errorf("opening fs repo: %w", err)
 		}
 
+		ctx := context.TODO()
+
 		exists, err := r.Exists()
 		if err != nil {
 			return err
@@ -113,7 +118,7 @@ var importObjectCmd = &cli.Command{
 		}
 		defer lr.Close() //nolint:errcheck
 
-		bs, err := lr.Blockstore(repo.BlockstoreChain)
+		bs, err := lr.Blockstore(ctx, repo.BlockstoreChain)
 		if err != nil {
 			return fmt.Errorf("failed to open blockstore: %w", err)
 		}

diff --git a/cmd/lotus-shed/pruning.go b/cmd/lotus-shed/pruning.go
@@ -131,7 +131,7 @@ var stateTreePruneCmd = &cli.Command{
 
 		defer lkrepo.Close() //nolint:errcheck
 
-		bs, err := lkrepo.Blockstore(repo.BlockstoreChain)
+		bs, err := lkrepo.Blockstore(ctx, repo.BlockstoreChain)
 		if err != nil {
 			return fmt.Errorf("failed to open blockstore: %w", err)
 		}
@@ -151,7 +151,7 @@ var stateTreePruneCmd = &cli.Command{
 			return fmt.Errorf("only badger blockstores are supported")
 		}
 
-		mds, err := lkrepo.Datastore("/metadata")
+		mds, err := lkrepo.Datastore(context.Background(), "/metadata")
 		if err != nil {
 			return err
 		}

diff --git a/cmd/lotus-storage-miner/init.go b/cmd/lotus-storage-miner/init.go
@@ -417,7 +417,7 @@ func storageMinerInit(ctx context.Context, cctx *cli.Context, api lapi.FullNode,
 		return xerrors.Errorf("peer ID from private key: %w", err)
 	}
 
-	mds, err := lr.Datastore("/metadata")
+	mds, err := lr.Datastore(context.TODO(), "/metadata")
 	if err != nil {
 		return err
 	}

diff --git a/cmd/lotus-storage-miner/init_restore.go b/cmd/lotus-storage-miner/init_restore.go
@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	"encoding/json"
 	"io/ioutil"
 	"os"
@@ -190,7 +191,7 @@ var initRestoreCmd = &cli.Command{
 
 		log.Info("Restoring metadata backup")
 
-		mds, err := lr.Datastore("/metadata")
+		mds, err := lr.Datastore(context.TODO(), "/metadata")
 		if err != nil {
 			return err
 		}

diff --git a/cmd/lotus-wallet/main.go b/cmd/lotus-wallet/main.go
@@ -118,7 +118,7 @@ var runCmd = &cli.Command{
 
 		var w api.WalletAPI = lw
 		if cctx.Bool("ledger") {
-			ds, err := lr.Datastore("/metadata")
+			ds, err := lr.Datastore(context.Background(), "/metadata")
 			if err != nil {
 				return err
 			}

diff --git a/cmd/lotus/backup.go b/cmd/lotus/backup.go
@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	"os"
 
 	dstore "github.com/ipfs/go-datastore"
@@ -87,7 +88,7 @@ func restore(cctx *cli.Context, r repo.Repo) error {
 
 	log.Info("Restoring metadata backup")
 
-	mds, err := lr.Datastore("/metadata")
+	mds, err := lr.Datastore(context.TODO(), "/metadata")
 	if err != nil {
 		return err
 	}

diff --git a/cmd/lotus/daemon.go b/cmd/lotus/daemon.go
@@ -254,7 +254,7 @@ var DaemonCmd = &cli.Command{
 				issnapshot = true
 			}
 
-			if err := ImportChain(r, chainfile, issnapshot); err != nil {
+			if err := ImportChain(ctx, r, chainfile, issnapshot); err != nil {
 				return err
 			}
 			if cctx.Bool("halt-after-import") {
@@ -389,7 +389,7 @@ func importKey(ctx context.Context, api api.FullNode, f string) error {
 	return nil
 }
 
-func ImportChain(r repo.Repo, fname string, snapshot bool) (err error) {
+func ImportChain(ctx context.Context, r repo.Repo, fname string, snapshot bool) (err error) {
 	var rd io.Reader
 	var l int64
 	if strings.HasPrefix(fname, "http://") || strings.HasPrefix(fname, "https://") {
@@ -432,12 +432,12 @@ func ImportChain(r repo.Repo, fname string, snapshot bool) (err error) {
 	}
 	defer lr.Close() //nolint:errcheck
 
-	bs, err := lr.Blockstore(repo.BlockstoreChain)
+	bs, err := lr.Blockstore(ctx, repo.BlockstoreChain)
 	if err != nil {
 		return xerrors.Errorf("failed to open blockstore: %w", err)
 	}
 
-	mds, err := lr.Datastore("/metadata")
+	mds, err := lr.Datastore(context.TODO(), "/metadata")
 	if err != nil {
 		return err
 	}
@@ -473,7 +473,7 @@ func ImportChain(r repo.Repo, fname string, snapshot bool) (err error) {
 		return xerrors.Errorf("flushing validation cache failed: %w", err)
 	}
 
-	gb, err := cst.GetTipsetByHeight(context.TODO(), 0, ts, true)
+	gb, err := cst.GetTipsetByHeight(ctx, 0, ts, true)
 	if err != nil {
 		return err
 	}
@@ -487,13 +487,13 @@ func ImportChain(r repo.Repo, fname string, snapshot bool) (err error) {
 
 	if !snapshot {
 		log.Infof("validating imported chain...")
-		if err := stm.ValidateChain(context.TODO(), ts); err != nil {
+		if err := stm.ValidateChain(ctx, ts); err != nil {
 			return xerrors.Errorf("chain validation failed: %w", err)
 		}
 	}
 
 	log.Infof("accepting %s as new head", ts.Cids())
-	if err := cst.ForceHeadSilent(context.Background(), ts); err != nil {
+	if err := cst.ForceHeadSilent(ctx, ts); err != nil {
 		return err
 	}
 

diff --git a/node/modules/chain.go b/node/modules/chain.go
@@ -77,7 +77,7 @@ func MessagePool(lc fx.Lifecycle, sm *stmgr.StateManager, ps *pubsub.PubSub, ds
 }
 
 func ChainRawBlockstore(lc fx.Lifecycle, mctx helpers.MetricsCtx, r repo.LockedRepo) (dtypes.ChainRawBlockstore, error) {
-	bs, err := r.Blockstore(repo.BlockstoreChain)
+	bs, err := r.Blockstore(helpers.LifecycleCtx(mctx, lc), repo.BlockstoreChain)
 	if err != nil {
 		return nil, err
 	}

diff --git a/node/modules/client.go b/node/modules/client.go
@@ -39,6 +39,7 @@ import (
 	"github.com/filecoin-project/lotus/node/impl/full"
 	payapi "github.com/filecoin-project/lotus/node/impl/paych"
 	"github.com/filecoin-project/lotus/node/modules/dtypes"
+	"github.com/filecoin-project/lotus/node/modules/helpers"
 	"github.com/filecoin-project/lotus/node/repo"
 	"github.com/filecoin-project/lotus/node/repo/importmgr"
 	"github.com/filecoin-project/lotus/node/repo/retrievalstoremgr"
@@ -78,8 +79,9 @@ func HandleMigrateClientFunds(lc fx.Lifecycle, ds dtypes.MetadataDS, wallet full
 	})
 }
 
-func ClientMultiDatastore(lc fx.Lifecycle, r repo.LockedRepo) (dtypes.ClientMultiDstore, error) {
-	ds, err := r.Datastore("/client")
+func ClientMultiDatastore(lc fx.Lifecycle, mctx helpers.MetricsCtx, r repo.LockedRepo) (dtypes.ClientMultiDstore, error) {
+	ctx := helpers.LifecycleCtx(mctx, lc)
+	ds, err := r.Datastore(ctx, "/client")
 	if err != nil {
 		return nil, xerrors.Errorf("getting datastore out of reop: %w", err)
 	}

diff --git a/node/modules/storage.go b/node/modules/storage.go
@@ -8,6 +8,7 @@ import (
 	"github.com/filecoin-project/lotus/chain/types"
 	"github.com/filecoin-project/lotus/lib/backupds"
 	"github.com/filecoin-project/lotus/node/modules/dtypes"
+	"github.com/filecoin-project/lotus/node/modules/helpers"
 	"github.com/filecoin-project/lotus/node/repo"
 )
 
@@ -27,8 +28,9 @@ func KeyStore(lr repo.LockedRepo) (types.KeyStore, error) {
 	return lr.KeyStore()
 }
 
-func Datastore(r repo.LockedRepo) (dtypes.MetadataDS, error) {
-	mds, err := r.Datastore("/metadata")
+func Datastore(lc fx.Lifecycle, mctx helpers.MetricsCtx, r repo.LockedRepo) (dtypes.MetadataDS, error) {
+	ctx := helpers.LifecycleCtx(mctx, lc)
+	mds, err := r.Datastore(ctx, "/metadata")
 	if err != nil {
 		return nil, err
 	}

diff --git a/node/modules/storageminer.go b/node/modules/storageminer.go
@@ -362,8 +362,9 @@ func NewProviderPieceStore(lc fx.Lifecycle, ds dtypes.MetadataDS) (dtypes.Provid
 	return ps, nil
 }
 
-func StagingMultiDatastore(lc fx.Lifecycle, r repo.LockedRepo) (dtypes.StagingMultiDstore, error) {
-	ds, err := r.Datastore("/staging")
+func StagingMultiDatastore(lc fx.Lifecycle, mctx helpers.MetricsCtx, r repo.LockedRepo) (dtypes.StagingMultiDstore, error) {
+	ctx := helpers.LifecycleCtx(mctx, lc)
+	ds, err := r.Datastore(ctx, "/staging")
 	if err != nil {
 		return nil, xerrors.Errorf("getting datastore out of reop: %w", err)
 	}
@@ -384,8 +385,9 @@ func StagingMultiDatastore(lc fx.Lifecycle, r repo.LockedRepo) (dtypes.StagingMu
 
 // StagingBlockstore creates a blockstore for staging blocks for a miner
 // in a storage deal, prior to sealing
-func StagingBlockstore(r repo.LockedRepo) (dtypes.StagingBlockstore, error) {
-	stagingds, err := r.Datastore("/staging")
+func StagingBlockstore(lc fx.Lifecycle, mctx helpers.MetricsCtx, r repo.LockedRepo) (dtypes.StagingBlockstore, error) {
+	ctx := helpers.LifecycleCtx(mctx, lc)
+	stagingds, err := r.Datastore(ctx, "/staging")
 	if err != nil {
 		return nil, err
 	}