update go-car/v2 to error instead of panic on misuse; make gen.

This was there to deal with migration IO workload. This is no
longer a problem with the new migration logic. The initialize-all
operation/command deals with concurrency.

For ongoing operation, miners can set the value they wish, but
we do not make a decision for them by default, because each
deployment can handle different IO loads.

CHANGELOG.md

@@ -1,10 +1,12 @@
 # Lotus changelog
 
-# 1.11.1-rc1 / 2021-07-27
+# 1.11.1-rc2 / 2021-08-01
 
-This is the first release candidate for the **highly-recommended** but optional Lotus v1.11.1 release that introduces many deal making and datastore improvements and new features along with other bug fixes. A more organized and detailed release note will be shared in the next few days with future RCs, highlights are:
-- [lotus-miner market subsystem](https://docs.filecoin.io/mine/lotus/split-markets-miners/#frontmatter-title)
-- [splistore](https://github.com/filecoin-project/lotus/blob/master/blockstore/splitstore/README.md)
+> Note: for discussion about this release, please comment [here](https://github.com/filecoin-project/lotus/discussions/6904)
+
+This is the second release candidate for the **highly-recommended** but optional Lotus v1.11.1 release that introduces many deal making and datastore improvements and new features along with other bug fixes. A more organized and detailed release note will be shared in the next few days with future RCs, highlights are:
+- **Experimental** [lotus-miner market subsystem](https://docs.filecoin.io/mine/lotus/split-markets-miners/#frontmatter-title)
+- **Under Testing** [splistore](https://github.com/filecoin-project/lotus/blob/master/blockstore/splitstore/README.md)
 
 - github.com/filecoin-project/lotus:
   - Merge branch 'releases' into release/v1.11.1

api/api_storage.go

@@ -13,13 +13,14 @@ import (
 
 	"github.com/filecoin-project/go-address"
 	datatransfer "github.com/filecoin-project/go-data-transfer"
-	"github.com/filecoin-project/go-fil-markets/piecestore"
-	"github.com/filecoin-project/go-fil-markets/retrievalmarket"
-	"github.com/filecoin-project/go-fil-markets/storagemarket"
 	"github.com/filecoin-project/go-state-types/abi"
 	"github.com/filecoin-project/specs-actors/v2/actors/builtin/market"
 	"github.com/filecoin-project/specs-storage/storage"
 
+	"github.com/filecoin-project/go-fil-markets/piecestore"
+	"github.com/filecoin-project/go-fil-markets/retrievalmarket"
+	"github.com/filecoin-project/go-fil-markets/storagemarket"
+
 	"github.com/filecoin-project/lotus/chain/types"
 	"github.com/filecoin-project/lotus/extern/sector-storage/fsutil"
 	"github.com/filecoin-project/lotus/extern/sector-storage/stores"
@@ -166,6 +167,48 @@ type StorageMiner interface {
 	MarketPendingDeals(ctx context.Context) (PendingDealInfo, error)                                                             //perm:write
 	MarketPublishPendingDeals(ctx context.Context) error                                                                         //perm:admin
 
+	// DagstoreListShards returns information about all shards known to the
+	// DAG store. Only available on nodes running the markets subsystem.
+	DagstoreListShards(ctx context.Context) ([]DagstoreShardInfo, error) //perm:read
+
+	// DagstoreInitializeShard initializes an uninitialized shard.
+	//
+	// Initialization consists of fetching the shard's data (deal payload) from
+	// the storage subsystem, generating an index, and persisting the index
+	// to facilitate later retrievals, and/or to publish to external sources.
+	//
+	// This operation is intended to complement the initial migration. The
+	// migration registers a shard for every unique piece CID, with lazy
+	// initialization. Thus, shards are not initialized immediately to avoid
+	// IO activity competing with proving. Instead, shard are initialized
+	// when first accessed. This method forces the initialization of a shard by
+	// accessing it and immediately releasing it. This is useful to warm up the
+	// cache to facilitate subsequent retrievals, and to generate the indexes
+	// to publish them externally.
+	//
+	// This operation fails if the shard is not in ShardStateNew state.
+	// It blocks until initialization finishes.
+	DagstoreInitializeShard(ctx context.Context, key string) error //perm:write
+
+	// DagstoreRecoverShard attempts to recover a failed shard.
+	//
+	// This operation fails if the shard is not in ShardStateErrored state.
+	// It blocks until recovery finishes. If recovery failed, it returns the
+	// error.
+	DagstoreRecoverShard(ctx context.Context, key string) error //perm:write
+
+	// DagstoreInitializeAll initializes all uninitialized shards in bulk,
+	// according to the policy passed in the parameters.
+	//
+	// It is recommended to set a maximum concurrency to avoid extreme
+	// IO pressure if the storage subsystem has a large amount of deals.
+	//
+	// It returns a stream of events to report progress.
+	DagstoreInitializeAll(ctx context.Context, params DagstoreInitializeAllParams) (<-chan DagstoreInitializeAllEvent, error) //perm:write
+
+	// DagstoreGC runs garbage collection on the DAG store.
+	DagstoreGC(ctx context.Context) ([]DagstoreShardResult, error) //perm:admin
+
 	// RuntimeSubsystems returns the subsystems that are enabled
 	// in this instance.
 	RuntimeSubsystems(ctx context.Context) (MinerSubsystems, error) //perm:read
@@ -336,3 +379,33 @@ type DealSchedule struct {
 	StartEpoch abi.ChainEpoch
 	EndEpoch   abi.ChainEpoch
 }
+
+// DagstoreShardInfo is the serialized form of dagstore.DagstoreShardInfo that
+// we expose through JSON-RPC to avoid clients having to depend on the
+// dagstore lib.
+type DagstoreShardInfo struct {
+	Key   string
+	State string
+	Error string
+}
+
+// DagstoreShardResult enumerates results per shard.
+type DagstoreShardResult struct {
+	Key     string
+	Success bool
+	Error   string
+}
+
+type DagstoreInitializeAllParams struct {
+	MaxConcurrency int
+}
+
+// DagstoreInitializeAllEvent represents an initialization event.
+type DagstoreInitializeAllEvent struct {
+	Key     string
+	Event   string // "start", "end"
+	Success bool
+	Error   string
+	Total   int
+	Current int
+}

api/api_test.go

@@ -76,7 +76,7 @@ func TestReturnTypes(t *testing.T) {
 						seen[typ] = struct{}{}
 
 						if typ.Kind() == reflect.Interface && typ != bareIface && !typ.Implements(jmarsh) {
-							t.Error("methods can't return interfaces", m.Name)
+							t.Error("methods can't return interfaces or struct types not implementing json.Marshaller", m.Name)
 						}
 
 						switch typ.Kind() {

api/docgen/docgen.go

@@ -14,7 +14,6 @@ import (
 	"github.com/filecoin-project/go-address"
 	"github.com/filecoin-project/go-bitfield"
 	"github.com/filecoin-project/go-multistore"
-	"github.com/filecoin-project/lotus/node/repo/importmgr"
 	"github.com/google/uuid"
 	"github.com/ipfs/go-cid"
 	"github.com/ipfs/go-filestore"
@@ -25,6 +24,8 @@ import (
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	"github.com/multiformats/go-multiaddr"
 
+	"github.com/filecoin-project/lotus/node/repo/importmgr"
+
 	datatransfer "github.com/filecoin-project/go-data-transfer"
 	filestore2 "github.com/filecoin-project/go-fil-markets/filestore"
 	"github.com/filecoin-project/go-fil-markets/retrievalmarket"
@@ -275,6 +276,15 @@ func init() {
 		api.SubsystemSectorStorage,
 		api.SubsystemMarkets,
 	})
+	addExample(api.DagstoreShardResult{
+		Key:   "baga6ea4seaqecmtz7iak33dsfshi627abz4i4665dfuzr3qfs4bmad6dx3iigdq",
+		Error: "<error>",
+	})
+	addExample(api.DagstoreShardInfo{
+		Key:   "baga6ea4seaqecmtz7iak33dsfshi627abz4i4665dfuzr3qfs4bmad6dx3iigdq",
+		State: "ShardStateAvailable",
+		Error: "<error>",
+	})
 }
 
 func GetAPIType(name, pkg string) (i interface{}, t reflect.Type, permStruct []reflect.Type) {

api/proxy_gen.go

@@ -603,6 +603,16 @@ type StorageMinerStruct struct {
 
 		CreateBackup func(p0 context.Context, p1 string) error `perm:"admin"`
 
+		DagstoreGC func(p0 context.Context) ([]DagstoreShardResult, error) `perm:"admin"`
+
+		DagstoreInitializeAll func(p0 context.Context, p1 DagstoreInitializeAllParams) (<-chan DagstoreInitializeAllEvent, error) `perm:"write"`
+
+		DagstoreInitializeShard func(p0 context.Context, p1 string) error `perm:"write"`
+
+		DagstoreListShards func(p0 context.Context) ([]DagstoreShardInfo, error) `perm:"read"`
+
+		DagstoreRecoverShard func(p0 context.Context, p1 string) error `perm:"write"`
+
 		DealsConsiderOfflineRetrievalDeals func(p0 context.Context) (bool, error) `perm:"admin"`
 
 		DealsConsiderOfflineStorageDeals func(p0 context.Context) (bool, error) `perm:"admin"`
@@ -3569,6 +3579,61 @@ func (s *StorageMinerStub) CreateBackup(p0 context.Context, p1 string) error {
 	return ErrNotSupported
 }
 
+func (s *StorageMinerStruct) DagstoreGC(p0 context.Context) ([]DagstoreShardResult, error) {
+	if s.Internal.DagstoreGC == nil {
+		return *new([]DagstoreShardResult), ErrNotSupported
+	}
+	return s.Internal.DagstoreGC(p0)
+}
+
+func (s *StorageMinerStub) DagstoreGC(p0 context.Context) ([]DagstoreShardResult, error) {
+	return *new([]DagstoreShardResult), ErrNotSupported
+}
+
+func (s *StorageMinerStruct) DagstoreInitializeAll(p0 context.Context, p1 DagstoreInitializeAllParams) (<-chan DagstoreInitializeAllEvent, error) {
+	if s.Internal.DagstoreInitializeAll == nil {
+		return nil, ErrNotSupported
+	}
+	return s.Internal.DagstoreInitializeAll(p0, p1)
+}
+
+func (s *StorageMinerStub) DagstoreInitializeAll(p0 context.Context, p1 DagstoreInitializeAllParams) (<-chan DagstoreInitializeAllEvent, error) {
+	return nil, ErrNotSupported
+}
+
+func (s *StorageMinerStruct) DagstoreInitializeShard(p0 context.Context, p1 string) error {
+	if s.Internal.DagstoreInitializeShard == nil {
+		return ErrNotSupported
+	}
+	return s.Internal.DagstoreInitializeShard(p0, p1)
+}
+
+func (s *StorageMinerStub) DagstoreInitializeShard(p0 context.Context, p1 string) error {
+	return ErrNotSupported
+}
+
+func (s *StorageMinerStruct) DagstoreListShards(p0 context.Context) ([]DagstoreShardInfo, error) {
+	if s.Internal.DagstoreListShards == nil {
+		return *new([]DagstoreShardInfo), ErrNotSupported
+	}
+	return s.Internal.DagstoreListShards(p0)
+}
+
+func (s *StorageMinerStub) DagstoreListShards(p0 context.Context) ([]DagstoreShardInfo, error) {
+	return *new([]DagstoreShardInfo), ErrNotSupported
+}
+
+func (s *StorageMinerStruct) DagstoreRecoverShard(p0 context.Context, p1 string) error {
+	if s.Internal.DagstoreRecoverShard == nil {
+		return ErrNotSupported
+	}
+	return s.Internal.DagstoreRecoverShard(p0, p1)
+}
+
+func (s *StorageMinerStub) DagstoreRecoverShard(p0 context.Context, p1 string) error {
+	return ErrNotSupported
+}
+
 func (s *StorageMinerStruct) DealsConsiderOfflineRetrievalDeals(p0 context.Context) (bool, error) {
 	if s.Internal.DealsConsiderOfflineRetrievalDeals == nil {
 		return false, ErrNotSupported

build/version.go

@@ -34,7 +34,7 @@ func buildType() string {
 }
 
 // BuildVersion is the local build version, set by build system
-const BuildVersion = "1.11.1-rc1"
+const BuildVersion = "1.11.1-m1.3.4"
 
 func UserVersion() string {
 	if os.Getenv("LOTUS_VERSION_IGNORE_COMMIT") == "1" {

cli/util/api.go

@@ -2,6 +2,7 @@ package cliutil
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"net/http"
 	"net/url"
@@ -142,6 +143,15 @@ func GetAPIInfo(ctx *cli.Context, t repo.RepoType) (APIInfo, error) {
 			return APIInfo{}, xerrors.Errorf("could not open repo at path: %s; %w", p, err)
 		}
 
+		exists, err := r.Exists()
+		if err != nil {
+			return APIInfo{}, xerrors.Errorf("repo.Exists returned an error: %w", err)
+		}
+
+		if !exists {
+			return APIInfo{}, errors.New("repo directory does not exist. Make sure your configuration is correct")
+		}
+
 		ma, err := r.APIEndpoint()
 		if err != nil {
 			return APIInfo{}, xerrors.Errorf("could not get api endpoint: %w", err)

cli/wait.go

@@ -22,7 +22,7 @@ var WaitApiCmd = &cli.Command{
 
 			ctx := ReqContext(cctx)
 
-			_, err = api.ID(ctx)
+			_, err = api.Version(ctx)
 			if err != nil {
 				return err
 			}

cmd/lotus-miner/dagstore.go

@@ -0,0 +1,209 @@
+package main
+
+import (
+	"fmt"
+	"os"
+
+	"github.com/fatih/color"
+	"github.com/urfave/cli/v2"
+
+	"github.com/filecoin-project/lotus/api"
+	lcli "github.com/filecoin-project/lotus/cli"
+	"github.com/filecoin-project/lotus/lib/tablewriter"
+)
+
+var dagstoreCmd = &cli.Command{
+	Name:  "dagstore",
+	Usage: "Manage the dagstore on the markets subsystem",
+	Subcommands: []*cli.Command{
+		dagstoreListShardsCmd,
+		dagstoreInitializeShardCmd,
+		dagstoreRecoverShardCmd,
+		dagstoreInitializeAllCmd,
+		dagstoreGcCmd,
+	},
+}
+
+var dagstoreListShardsCmd = &cli.Command{
+	Name:  "list-shards",
+	Usage: "List all shards known to the dagstore, with their current status",
+	Action: func(cctx *cli.Context) error {
+		marketsApi, closer, err := lcli.GetMarketsAPI(cctx)
+		if err != nil {
+			return err
+		}
+		defer closer()
+
+		ctx := lcli.ReqContext(cctx)
+
+		shards, err := marketsApi.DagstoreListShards(ctx)
+		if err != nil {
+			return err
+		}
+
+		if len(shards) == 0 {
+			return nil
+		}
+
+		tw := tablewriter.New(
+			tablewriter.Col("Key"),
+			tablewriter.Col("State"),
+			tablewriter.Col("Error"),
+		)
+
+		colors := map[string]color.Attribute{
+			"ShardStateAvailable": color.FgGreen,
+			"ShardStateServing":   color.FgBlue,
+			"ShardStateErrored":   color.FgRed,
+			"ShardStateNew":       color.FgYellow,
+		}
+
+		for _, s := range shards {
+			m := map[string]interface{}{
+				"Key": s.Key,
+				"State": func() string {
+					if c, ok := colors[s.State]; ok {
+						return color.New(c).Sprint(s.State)
+					}
+					return s.State
+				}(),
+				"Error": s.Error,
+			}
+			tw.Write(m)
+		}
+
+		return tw.Flush(os.Stdout)
+	},
+}
+
+var dagstoreInitializeShardCmd = &cli.Command{
+	Name:      "initialize-shard",
+	ArgsUsage: "[key]",
+	Usage:     "Initialize the specified shard",
+	Action: func(cctx *cli.Context) error {
+		if cctx.NArg() != 1 {
+			return fmt.Errorf("must provide a single shard key")
+		}
+
+		marketsApi, closer, err := lcli.GetMarketsAPI(cctx)
+		if err != nil {
+			return err
+		}
+		defer closer()
+
+		ctx := lcli.ReqContext(cctx)
+
+		return marketsApi.DagstoreInitializeShard(ctx, cctx.Args().First())
+	},
+}
+
+var dagstoreRecoverShardCmd = &cli.Command{
+	Name:      "recover-shard",
+	ArgsUsage: "[key]",
+	Usage:     "Attempt to recover a shard in errored state",
+	Action: func(cctx *cli.Context) error {
+		if cctx.NArg() != 1 {
+			return fmt.Errorf("must provide a single shard key")
+		}
+
+		marketsApi, closer, err := lcli.GetMarketsAPI(cctx)
+		if err != nil {
+			return err
+		}
+		defer closer()
+
+		ctx := lcli.ReqContext(cctx)
+
+		return marketsApi.DagstoreRecoverShard(ctx, cctx.Args().First())
+	},
+}
+
+var dagstoreInitializeAllCmd = &cli.Command{
+	Name:  "initialize-all",
+	Usage: "Initialize all uninitialized shards, streaming results as they're produced",
+	Flags: []cli.Flag{
+		&cli.UintFlag{
+			Name:     "concurrency",
+			Usage:    "maximum shards to initialize concurrently at a time; use 0 for unlimited",
+			Required: true,
+		},
+	},
+	Action: func(cctx *cli.Context) error {
+		concurrency := cctx.Uint("concurrency")
+
+		marketsApi, closer, err := lcli.GetMarketsAPI(cctx)
+		if err != nil {
+			return err
+		}
+		defer closer()
+
+		ctx := lcli.ReqContext(cctx)
+
+		params := api.DagstoreInitializeAllParams{
+			MaxConcurrency: int(concurrency),
+		}
+
+		ch, err := marketsApi.DagstoreInitializeAll(ctx, params)
+		if err != nil {
+			return err
+		}
+
+		for {
+			select {
+			case evt, ok := <-ch:
+				if !ok {
+					return nil
+				}
+				_, _ = fmt.Fprint(os.Stdout, color.New(color.BgHiBlack).Sprintf("(%d/%d)", evt.Current, evt.Total))
+				_, _ = fmt.Fprint(os.Stdout, " ")
+				if evt.Event == "start" {
+					_, _ = fmt.Fprintln(os.Stdout, evt.Key, color.New(color.Reset).Sprint("STARTING"))
+				} else {
+					if evt.Success {
+						_, _ = fmt.Fprintln(os.Stdout, evt.Key, color.New(color.FgGreen).Sprint("SUCCESS"))
+					} else {
+						_, _ = fmt.Fprintln(os.Stdout, evt.Key, color.New(color.FgRed).Sprint("ERROR"), evt.Error)
+					}
+				}
+
+			case <-ctx.Done():
+				return fmt.Errorf("aborted")
+			}
+		}
+	},
+}
+
+var dagstoreGcCmd = &cli.Command{
+	Name:  "gc",
+	Usage: "Garbage collect the dagstore",
+
+	Action: func(cctx *cli.Context) error {
+		marketsApi, closer, err := lcli.GetMarketsAPI(cctx)
+		if err != nil {
+			return err
+		}
+		defer closer()
+
+		ctx := lcli.ReqContext(cctx)
+
+		collected, err := marketsApi.DagstoreGC(ctx)
+		if err != nil {
+			return err
+		}
+
+		if len(collected) == 0 {
+			_, _ = fmt.Fprintln(os.Stdout, "no shards collected")
+			return nil
+		}
+
+		for _, e := range collected {
+			if e.Error == "" {
+				_, _ = fmt.Fprintln(os.Stdout, e.Key, color.New(color.FgGreen).Sprint("SUCCESS"))
+			} else {
+				_, _ = fmt.Fprintln(os.Stdout, e.Key, color.New(color.FgRed).Sprint("ERROR"), e.Error)
+			}
+		}
+
+		return nil
+	},
+}

cmd/lotus-miner/main.go

@@ -5,13 +5,15 @@ import (
 	"fmt"
 
 	"github.com/fatih/color"
-	cliutil "github.com/filecoin-project/lotus/cli/util"
 	logging "github.com/ipfs/go-log/v2"
 	"github.com/urfave/cli/v2"
 	"go.opencensus.io/trace"
 	"golang.org/x/xerrors"
 
+	cliutil "github.com/filecoin-project/lotus/cli/util"
+
 	"github.com/filecoin-project/go-address"
+
 	"github.com/filecoin-project/lotus/api"
 	"github.com/filecoin-project/lotus/build"
 	lcli "github.com/filecoin-project/lotus/cli"
@@ -46,6 +48,7 @@ func main() {
 		lcli.WithCategory("market", storageDealsCmd),
 		lcli.WithCategory("market", retrievalDealsCmd),
 		lcli.WithCategory("market", dataTransfersCmd),
+		lcli.WithCategory("market", dagstoreCmd),
 		lcli.WithCategory("storage", sectorsCmd),
 		lcli.WithCategory("storage", provingCmd),
 		lcli.WithCategory("storage", storageCmd),

cmd/lotus-miner/market.go

@@ -591,7 +591,7 @@ var setSealDurationCmd = &cli.Command{
 	Usage:     "Set the expected time, in minutes, that you expect sealing sectors to take. Deals that start before this duration will be rejected.",
 	ArgsUsage: "<minutes>",
 	Action: func(cctx *cli.Context) error {
-		nodeApi, closer, err := lcli.GetStorageMinerAPI(cctx)
+		nodeApi, closer, err := lcli.GetMarketsAPI(cctx)
 		if err != nil {
 			return err
 		}

cmd/lotus-miner/pieces.go

@@ -26,7 +26,7 @@ var piecesListPiecesCmd = &cli.Command{
 	Name:  "list-pieces",
 	Usage: "list registered pieces",
 	Action: func(cctx *cli.Context) error {
-		nodeApi, closer, err := lcli.GetStorageMinerAPI(cctx)
+		nodeApi, closer, err := lcli.GetMarketsAPI(cctx)
 		if err != nil {
 			return err
 		}
@@ -49,7 +49,7 @@ var piecesListCidInfosCmd = &cli.Command{
 	Name:  "list-cids",
 	Usage: "list registered payload CIDs",
 	Action: func(cctx *cli.Context) error {
-		nodeApi, closer, err := lcli.GetStorageMinerAPI(cctx)
+		nodeApi, closer, err := lcli.GetMarketsAPI(cctx)
 		if err != nil {
 			return err
 		}
@@ -76,7 +76,7 @@ var piecesInfoCmd = &cli.Command{
 			return lcli.ShowHelp(cctx, fmt.Errorf("must specify piece cid"))
 		}
 
-		nodeApi, closer, err := lcli.GetStorageMinerAPI(cctx)
+		nodeApi, closer, err := lcli.GetMarketsAPI(cctx)
 		if err != nil {
 			return err
 		}
@@ -111,7 +111,7 @@ var piecesCidInfoCmd = &cli.Command{
 			return lcli.ShowHelp(cctx, fmt.Errorf("must specify payload cid"))
 		}
 
-		nodeApi, closer, err := lcli.GetStorageMinerAPI(cctx)
+		nodeApi, closer, err := lcli.GetMarketsAPI(cctx)
 		if err != nil {
 			return err
 		}

cmd/lotus-miner/sectors.go

@@ -396,7 +396,7 @@ var sectorsRefsCmd = &cli.Command{
 	Name:  "refs",
 	Usage: "List References to sectors",
 	Action: func(cctx *cli.Context) error {
-		nodeApi, closer, err := lcli.GetStorageMinerAPI(cctx)
+		nodeApi, closer, err := lcli.GetMarketsAPI(cctx)
 		if err != nil {
 			return err
 		}

documentation/en/api-v0-methods-miner.md

@@ -18,6 +18,12 @@
   * [ComputeProof](#ComputeProof)
 * [Create](#Create)
   * [CreateBackup](#CreateBackup)
+* [Dagstore](#Dagstore)
+  * [DagstoreGC](#DagstoreGC)
+  * [DagstoreInitializeAll](#DagstoreInitializeAll)
+  * [DagstoreInitializeShard](#DagstoreInitializeShard)
+  * [DagstoreListShards](#DagstoreListShards)
+  * [DagstoreRecoverShard](#DagstoreRecoverShard)
 * [Deals](#Deals)
   * [DealsConsiderOfflineRetrievalDeals](#DealsConsiderOfflineRetrievalDeals)
   * [DealsConsiderOfflineStorageDeals](#DealsConsiderOfflineStorageDeals)
@@ -345,6 +351,113 @@ Inputs:
 
 Response: `{}`
 
+## Dagstore
+
+
+### DagstoreGC
+DagstoreGC runs garbage collection on the DAG store.
+
+
+Perms: admin
+
+Inputs: `null`
+
+Response: `null`
+
+### DagstoreInitializeAll
+DagstoreInitializeAll initializes all uninitialized shards in bulk,
+according to the policy passed in the parameters.
+
+It is recommended to set a maximum concurrency to avoid extreme
+IO pressure if the storage subsystem has a large amount of deals.
+
+It returns a stream of events to report progress.
+
+
+Perms: write
+
+Inputs:
+```json
+[
+  {
+    "MaxConcurrency": 123
+  }
+]
+```
+
+Response:
+```json
+{
+  "Key": "string value",
+  "Event": "string value",
+  "Success": true,
+  "Error": "string value",
+  "Total": 123,
+  "Current": 123
+}
+```
+
+### DagstoreInitializeShard
+DagstoreInitializeShard initializes an uninitialized shard.
+
+Initialization consists of fetching the shard's data (deal payload) from
+the storage subsystem, generating an index, and persisting the index
+to facilitate later retrievals, and/or to publish to external sources.
+
+This operation is intended to complement the initial migration. The
+migration registers a shard for every unique piece CID, with lazy
+initialization. Thus, shards are not initialized immediately to avoid
+IO activity competing with proving. Instead, shard are initialized
+when first accessed. This method forces the initialization of a shard by
+accessing it and immediately releasing it. This is useful to warm up the
+cache to facilitate subsequent retrievals, and to generate the indexes
+to publish them externally.
+
+This operation fails if the shard is not in ShardStateNew state.
+It blocks until initialization finishes.
+
+
+Perms: write
+
+Inputs:
+```json
+[
+  "string value"
+]
+```
+
+Response: `{}`
+
+### DagstoreListShards
+DagstoreListShards returns information about all shards known to the
+DAG store. Only available on nodes running the markets subsystem.
+
+
+Perms: read
+
+Inputs: `null`
+
+Response: `null`
+
+### DagstoreRecoverShard
+DagstoreRecoverShard attempts to recover a failed shard.
+
+This operation fails if the shard is not in ShardStateErrored state.
+It blocks until recovery finishes. If recovery failed, it returns the
+error.
+
+
+Perms: write
+
+Inputs:
+```json
+[
+  "string value"
+]
+```
+
+Response: `{}`
+
 ## Deals
 
 
@@ -709,7 +822,7 @@ Response:
     "ID": 3
   },
   "SectorNumber": 9,
-  "CARv2FilePath": "string value"
+  "InboundCAR": "string value"
 }
 ```
 

documentation/en/cli-lotus-miner.md

@@ -7,7 +7,7 @@ USAGE:
    lotus-miner [global options] command [command options] [arguments...]
 
 VERSION:
-   1.11.1-rc1
+   1.11.1-m1.3.4
 
 COMMANDS:
    init     Initialize a lotus miner repo
@@ -29,6 +29,7 @@ COMMANDS:
      storage-deals    Manage storage deals and related configuration
      retrieval-deals  Manage retrieval deals and related configuration
      data-transfers   Manage data transfers
+     dagstore         Manage the dagstore on the markets subsystem
    NETWORK:
      net  Manage P2P Network
    RETRIEVAL:
@@ -1000,6 +1001,94 @@ OPTIONS:
    
 ```
 
+## lotus-miner dagstore
+```
+NAME:
+   lotus-miner dagstore - Manage the dagstore on the markets subsystem
+
+USAGE:
+   lotus-miner dagstore command [command options] [arguments...]
+
+COMMANDS:
+   list-shards       List all shards known to the dagstore, with their current status
+   initialize-shard  Initialize the specified shard
+   recover-shard     Attempt to recover a shard in errored state
+   initialize-all    Initialize all uninitialized shards, streaming results as they're produced
+   gc                Garbage collect the dagstore
+   help, h           Shows a list of commands or help for one command
+
+OPTIONS:
+   --help, -h     show help (default: false)
+   --version, -v  print the version (default: false)
+   
+```
+
+### lotus-miner dagstore list-shards
+```
+NAME:
+   lotus-miner dagstore list-shards - List all shards known to the dagstore, with their current status
+
+USAGE:
+   lotus-miner dagstore list-shards [command options] [arguments...]
+
+OPTIONS:
+   --help, -h  show help (default: false)
+   
+```
+
+### lotus-miner dagstore initialize-shard
+```
+NAME:
+   lotus-miner dagstore initialize-shard - Initialize the specified shard
+
+USAGE:
+   lotus-miner dagstore initialize-shard [command options] [key]
+
+OPTIONS:
+   --help, -h  show help (default: false)
+   
+```
+
+### lotus-miner dagstore recover-shard
+```
+NAME:
+   lotus-miner dagstore recover-shard - Attempt to recover a shard in errored state
+
+USAGE:
+   lotus-miner dagstore recover-shard [command options] [key]
+
+OPTIONS:
+   --help, -h  show help (default: false)
+   
+```
+
+### lotus-miner dagstore initialize-all
+```
+NAME:
+   lotus-miner dagstore initialize-all - Initialize all uninitialized shards, streaming results as they're produced
+
+USAGE:
+   lotus-miner dagstore initialize-all [command options] [arguments...]
+
+OPTIONS:
+   --concurrency value  maximum shards to initialize concurrently at a time; use 0 for unlimited (default: 0)
+   --help, -h           show help (default: false)
+   
+```
+
+### lotus-miner dagstore gc
+```
+NAME:
+   lotus-miner dagstore gc - Garbage collect the dagstore
+
+USAGE:
+   lotus-miner dagstore gc [command options] [arguments...]
+
+OPTIONS:
+   --help, -h  show help (default: false)
+   
+```
+
 ## lotus-miner net
 ```
 NAME:

documentation/en/cli-lotus-worker.md

@@ -7,7 +7,7 @@ USAGE:
    lotus-worker [global options] command [command options] [arguments...]
 
 VERSION:
-   1.11.1-rc1
+   1.11.1-m1.3.4
 
 COMMANDS:
    run         Start lotus worker

documentation/en/cli-lotus.md

@@ -7,7 +7,7 @@ USAGE:
    lotus [global options] command [command options] [arguments...]
 
 VERSION:
-   1.11.1-rc1
+   1.11.1-m1.3.4
 
 COMMANDS:
    daemon   Start a lotus daemon process

go.mod

@@ -26,17 +26,17 @@ require (
 	github.com/elastic/gosigar v0.12.0
 	github.com/etclabscore/go-openrpc-reflect v0.0.36
 	github.com/fatih/color v1.9.0
-	github.com/filecoin-project/dagstore v0.3.2
+	github.com/filecoin-project/dagstore v0.4.2
 	github.com/filecoin-project/filecoin-ffi v0.30.4-0.20200910194244-f640612a1a1f
 	github.com/filecoin-project/go-address v0.0.5
 	github.com/filecoin-project/go-bitfield v0.2.4
 	github.com/filecoin-project/go-cbor-util v0.0.0-20191219014500-08c40a1e63a2
 	github.com/filecoin-project/go-commp-utils v0.1.1-0.20210427191551-70bf140d31c7
 	github.com/filecoin-project/go-crypto v0.0.0-20191218222705-effae4ea9f03
-	github.com/filecoin-project/go-data-transfer v1.7.2
+	github.com/filecoin-project/go-data-transfer v1.7.3
 	github.com/filecoin-project/go-fil-commcid v0.1.0
 	github.com/filecoin-project/go-fil-commp-hashhash v0.1.0
-	github.com/filecoin-project/go-fil-markets v1.6.3-0.20210729173742-a44f98a4f8c1
+	github.com/filecoin-project/go-fil-markets v1.6.3-0.20210806151421-967d8717c917
 	github.com/filecoin-project/go-jsonrpc v0.1.4-0.20210217175800-45ea43ac2bec
 	github.com/filecoin-project/go-multistore v0.0.3
 	github.com/filecoin-project/go-padreader v0.0.0-20210723183308-812a16dc01b1
@@ -56,7 +56,7 @@ require (
 	github.com/gdamore/tcell/v2 v2.2.0
 	github.com/go-kit/kit v0.10.0
 	github.com/go-ole/go-ole v1.2.4 // indirect
-	github.com/golang/mock v1.5.0
+	github.com/golang/mock v1.6.0
 	github.com/google/uuid v1.2.0
 	github.com/gorilla/mux v1.7.4
 	github.com/gorilla/websocket v1.4.2
@@ -78,7 +78,7 @@ require (
 	github.com/ipfs/go-ds-pebble v0.0.2-0.20200921225637-ce220f8ac459
 	github.com/ipfs/go-filestore v1.0.0
 	github.com/ipfs/go-fs-lock v0.0.6
-	github.com/ipfs/go-graphsync v0.6.6
+	github.com/ipfs/go-graphsync v0.6.8
 	github.com/ipfs/go-ipfs-blockstore v1.0.4
 	github.com/ipfs/go-ipfs-blocksutil v0.0.1
 	github.com/ipfs/go-ipfs-chunker v0.0.5
@@ -99,7 +99,7 @@ require (
 	github.com/ipfs/go-unixfs v0.2.6
 	github.com/ipfs/interface-go-ipfs-core v0.2.3
 	github.com/ipld/go-car v0.1.1-0.20201119040415-11b6074b6d4d
-	github.com/ipld/go-car/v2 v2.0.0-beta1.0.20210721090610-5a9d1b217d25
+	github.com/ipld/go-car/v2 v2.0.0-beta1.0.20210803143113-d0e44a62f0a3
 	github.com/ipld/go-ipld-prime v0.5.1-0.20201021195245-109253e8a018
 	github.com/kelseyhightower/envconfig v1.4.0
 	github.com/lib/pq v1.7.0

go.sum

@@ -256,9 +256,8 @@ github.com/fatih/color v1.8.0/go.mod h1:3l45GVGkyrnYNl9HoIjnp2NnNWvh6hLAqD8yTfGj
 github.com/fatih/color v1.9.0 h1:8xPHl4/q1VyqGIPif1F+1V3Y3lSmrq01EabUW3CoW5s=
 github.com/fatih/color v1.9.0/go.mod h1:eQcE1qtQxscV5RaZvpXrrb8Drkc3/DdQ+uUYCNjL+zU=
 github.com/fd/go-nat v1.0.0/go.mod h1:BTBu/CKvMmOMUPkKVef1pngt2WFH/lg7E6yQnulfp6E=
-github.com/filecoin-project/dagstore v0.3.1/go.mod h1:WY5OoLfnwISCk6eASSF927KKPqLPIlTwmG1qHpA08KY=
-github.com/filecoin-project/dagstore v0.3.2 h1:YgLW+0VpbjsWuRDUaPmjaS/KRz1cRbi0zAjZV/xfIwY=
-github.com/filecoin-project/dagstore v0.3.2/go.mod h1:WY5OoLfnwISCk6eASSF927KKPqLPIlTwmG1qHpA08KY=
+github.com/filecoin-project/dagstore v0.4.2 h1:Ae2+O1DhKCI1JbOZCBkqUksKYofdbRbjkS7OF0A6Jw0=
+github.com/filecoin-project/dagstore v0.4.2/go.mod h1:WY5OoLfnwISCk6eASSF927KKPqLPIlTwmG1qHpA08KY=
 github.com/filecoin-project/go-address v0.0.3/go.mod h1:jr8JxKsYx+lQlQZmF5i2U0Z+cGQ59wMIps/8YW/lDj8=
 github.com/filecoin-project/go-address v0.0.5 h1:SSaFT/5aLfPXycUlFyemoHYhRgdyXClXCyDdNJKPlDM=
 github.com/filecoin-project/go-address v0.0.5/go.mod h1:jr8JxKsYx+lQlQZmF5i2U0Z+cGQ59wMIps/8YW/lDj8=
@@ -279,8 +278,8 @@ github.com/filecoin-project/go-crypto v0.0.0-20191218222705-effae4ea9f03 h1:2pMX
 github.com/filecoin-project/go-crypto v0.0.0-20191218222705-effae4ea9f03/go.mod h1:+viYnvGtUTgJRdy6oaeF4MTFKAfatX071MPDPBL11EQ=
 github.com/filecoin-project/go-data-transfer v1.0.1/go.mod h1:UxvfUAY9v3ub0a21BSK9u3pB2aq30Y0KMsG+w9/ysyo=
 github.com/filecoin-project/go-data-transfer v1.7.0/go.mod h1:GLRr5BmLEqsLwXfiRDG7uJvph22KGL2M4iOuF8EINaU=
-github.com/filecoin-project/go-data-transfer v1.7.2 h1:iL3q5pxSloA7V2QucFofoVN3lquULz+Ml0KrNqMT5ZU=
-github.com/filecoin-project/go-data-transfer v1.7.2/go.mod h1:GLRr5BmLEqsLwXfiRDG7uJvph22KGL2M4iOuF8EINaU=
+github.com/filecoin-project/go-data-transfer v1.7.3 h1:2F3K1LmT6Q97+Bq9gBv/VVqFZ/0FHsRPqYDbCzrGaYg=
+github.com/filecoin-project/go-data-transfer v1.7.3/go.mod h1:Cbl9lzKOuAyyIxp1tE+VbV5Aix4bxzA7uJGA9wGM4fM=
 github.com/filecoin-project/go-ds-versioning v0.1.0 h1:y/X6UksYTsK8TLCI7rttCKEvl8btmWxyFMEeeWGUxIQ=
 github.com/filecoin-project/go-ds-versioning v0.1.0/go.mod h1:mp16rb4i2QPmxBnmanUx8i/XANp+PFCCJWiAb+VW4/s=
 github.com/filecoin-project/go-fil-commcid v0.0.0-20200716160307-8f644712406f/go.mod h1:Eaox7Hvus1JgPrL5+M3+h7aSPHc0cVqpSxA+TxIEpZQ=
@@ -290,8 +289,8 @@ github.com/filecoin-project/go-fil-commcid v0.1.0/go.mod h1:Eaox7Hvus1JgPrL5+M3+
 github.com/filecoin-project/go-fil-commp-hashhash v0.1.0 h1:imrrpZWEHRnNqqv0tN7LXep5bFEVOVmQWHJvl2mgsGo=
 github.com/filecoin-project/go-fil-commp-hashhash v0.1.0/go.mod h1:73S8WSEWh9vr0fDJVnKADhfIv/d6dCbAGaAGWbdJEI8=
 github.com/filecoin-project/go-fil-markets v1.0.5-0.20201113164554-c5eba40d5335/go.mod h1:AJySOJC00JRWEZzRG2KsfUnqEf5ITXxeX09BE9N4f9c=
-github.com/filecoin-project/go-fil-markets v1.6.3-0.20210729173742-a44f98a4f8c1 h1:xXgCsfEaA3wdof/N5mkD7PphDolhc9MAsIrO/QBH5Pg=
-github.com/filecoin-project/go-fil-markets v1.6.3-0.20210729173742-a44f98a4f8c1/go.mod h1:13+DUe7AaHekzgpQPbacdppRoqz0SyPlx48g0f/pRmA=
+github.com/filecoin-project/go-fil-markets v1.6.3-0.20210806151421-967d8717c917 h1:A7B4EShovtH4B8tMp6+Smst7B+H+CLSJ4tcjcS15ta4=
+github.com/filecoin-project/go-fil-markets v1.6.3-0.20210806151421-967d8717c917/go.mod h1:9DxT29OFLUJPswugvIbqD5Ab768pJQxkxbnTEFTI8lk=
 github.com/filecoin-project/go-hamt-ipld v0.1.5 h1:uoXrKbCQZ49OHpsTCkrThPNelC4W3LPEk0OrS/ytIBM=
 github.com/filecoin-project/go-hamt-ipld v0.1.5/go.mod h1:6Is+ONR5Cd5R6XZoCse1CWaXZc0Hdb/JeX+EQCQzX24=
 github.com/filecoin-project/go-hamt-ipld/v2 v2.0.0 h1:b3UDemBYN2HNfk3KOXNuxgTTxlWi3xVvbQP0IT38fvM=
@@ -435,8 +434,8 @@ github.com/golang/mock v1.3.1/go.mod h1:sBzyDLLjw3U8JLTeZvSv8jJB+tU5PVekmnlKIyFU
 github.com/golang/mock v1.4.0/go.mod h1:UOMv5ysSaYNkG+OFQykRIcU/QvvxJf3p21QfJ2Bt3cw=
 github.com/golang/mock v1.4.3/go.mod h1:UOMv5ysSaYNkG+OFQykRIcU/QvvxJf3p21QfJ2Bt3cw=
 github.com/golang/mock v1.4.4/go.mod h1:l3mdAwkq5BuhzHwde/uurv3sEJeZMXNpwsxVWU71h+4=
-github.com/golang/mock v1.5.0 h1:jlYHihg//f7RRwuPfptm04yp4s7O6Kw8EZiVYIGcH0g=
-github.com/golang/mock v1.5.0/go.mod h1:CWnOUgYIOo4TcNZ0wHX3YZCqsaM1I1Jvs6v3mP3KVu8=
+github.com/golang/mock v1.6.0 h1:ErTB+efbowRARo13NNdxyJji2egdxLGQhRaY+DUumQc=
+github.com/golang/mock v1.6.0/go.mod h1:p6yTPP+5HYm5mzsMV8JkE6ZKdX+/wYM6Hr+LicevLPs=
 github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/golang/protobuf v1.3.0/go.mod h1:Qd/q+1AKNOZr9uGQzbzCmRO6sUih6GTPZv6a1/R87v0=
 github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
@@ -640,8 +639,8 @@ github.com/ipfs/go-graphsync v0.1.0/go.mod h1:jMXfqIEDFukLPZHqDPp8tJMbHO9Rmeb9CE
 github.com/ipfs/go-graphsync v0.4.2/go.mod h1:/VmbZTUdUMTbNkgzAiCEucIIAU3BkLE2cZrDCVUhyi0=
 github.com/ipfs/go-graphsync v0.4.3/go.mod h1:mPOwDYv128gf8gxPFgXnz4fNrSYPsWyqisJ7ych+XDY=
 github.com/ipfs/go-graphsync v0.6.4/go.mod h1:5WyaeigpNdpiYQuW2vwpuecOoEfB4h747ZGEOKmAGTg=
-github.com/ipfs/go-graphsync v0.6.6 h1:In7jjzvSXlrAUz4OjN41lxYf/dzkf1bVeVxLpwKMRo8=
-github.com/ipfs/go-graphsync v0.6.6/go.mod h1:GdHT8JeuIZ0R4lSjFR16Oe4zPi5dXwKi9zR9ADVlcdk=
+github.com/ipfs/go-graphsync v0.6.8 h1:mgyPdBDPcgL8ujO132grQjP3rfQv+KN/riQzbmTVgg4=
+github.com/ipfs/go-graphsync v0.6.8/go.mod h1:GdHT8JeuIZ0R4lSjFR16Oe4zPi5dXwKi9zR9ADVlcdk=
 github.com/ipfs/go-hamt-ipld v0.1.1/go.mod h1:1EZCr2v0jlCnhpa+aZ0JZYp8Tt2w16+JJOAVz17YcDk=
 github.com/ipfs/go-ipfs-blockstore v0.0.1/go.mod h1:d3WClOmRQKFnJ0Jz/jj/zmksX0ma1gROTlovZKBmN08=
 github.com/ipfs/go-ipfs-blockstore v0.1.0/go.mod h1:5aD0AvHPi7mZc6Ci1WCAhiBQu2IsfTduLl+422H6Rqw=
@@ -760,8 +759,9 @@ github.com/ipld/go-car v0.1.0/go.mod h1:RCWzaUh2i4mOEkB3W45Vc+9jnS/M6Qay5ooytiBH
 github.com/ipld/go-car v0.1.1-0.20200923150018-8cdef32e2da4/go.mod h1:xrMEcuSq+D1vEwl+YAXsg/JfA98XGpXDwnkIL4Aimqw=
 github.com/ipld/go-car v0.1.1-0.20201119040415-11b6074b6d4d h1:iphSzTuPqyDgH7WUVZsdqUnQNzYgIblsVr1zhVNA33U=
 github.com/ipld/go-car v0.1.1-0.20201119040415-11b6074b6d4d/go.mod h1:2Gys8L8MJ6zkh1gktTSXreY63t4UbyvNp5JaudTyxHQ=
-github.com/ipld/go-car/v2 v2.0.0-beta1.0.20210721090610-5a9d1b217d25 h1:SeNfgKCuIhGJKpaSVvn8WX7uHxWeHvwVUq1MIh2KM2o=
 github.com/ipld/go-car/v2 v2.0.0-beta1.0.20210721090610-5a9d1b217d25/go.mod h1:I2ACeeg6XNBe5pdh5TaR7Ambhfa7If9KXxmXgZsYENU=
+github.com/ipld/go-car/v2 v2.0.0-beta1.0.20210803143113-d0e44a62f0a3 h1:TPOuaAmTWpzWFbGPsJ+QJsR6Qc+fnHKgvtA6Jy2A22w=
+github.com/ipld/go-car/v2 v2.0.0-beta1.0.20210803143113-d0e44a62f0a3/go.mod h1:I2ACeeg6XNBe5pdh5TaR7Ambhfa7If9KXxmXgZsYENU=
 github.com/ipld/go-ipld-prime v0.0.2-0.20191108012745-28a82f04c785/go.mod h1:bDDSvVz7vaK12FNvMeRYnpRFkSUPNQOiCYQezMD/P3w=
 github.com/ipld/go-ipld-prime v0.0.2-0.20200428162820-8b59dc292b8e/go.mod h1:uVIwe/u0H4VdKv3kaN1ck7uCb6yD9cFLS9/ELyXbsw8=
 github.com/ipld/go-ipld-prime v0.5.1-0.20200828233916-988837377a7f/go.mod h1:0xEgdD6MKbZ1vF0GC+YcR/C4SQCAlRuOjIJ2i0HxqzM=
@@ -1989,6 +1989,7 @@ golang.org/x/tools v0.0.0-20200827010519-17fd2f27a9e3/go.mod h1:njjCfa9FT2d7l9Bc
 golang.org/x/tools v0.0.0-20201112185108-eeaa07dd7696/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.1.0/go.mod h1:xkSsbof2nBLbhDlRMhhhyNLN/zl3eTqcnHD5viDpcZ0=
+golang.org/x/tools v0.1.1/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
 golang.org/x/tools v0.1.5 h1:ouewzE6p+/VEB31YYnTbEJdi8pFqKp4P4n85vwo3DHA=
 golang.org/x/tools v0.1.5/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=

markets/dagstore/blockstore.go

@@ -0,0 +1,33 @@
+package dagstore
+
+import (
+	"io"
+
+	blocks "github.com/ipfs/go-block-format"
+	"github.com/ipfs/go-cid"
+	bstore "github.com/ipfs/go-ipfs-blockstore"
+	"golang.org/x/xerrors"
+
+	"github.com/filecoin-project/dagstore"
+)
+
+// Blockstore promotes a dagstore.ReadBlockstore to a full closeable Blockstore,
+// stubbing out the write methods with erroring implementations.
+type Blockstore struct {
+	dagstore.ReadBlockstore
+	io.Closer
+}
+
+var _ bstore.Blockstore = (*Blockstore)(nil)
+
+func (b *Blockstore) DeleteBlock(c cid.Cid) error {
+	return xerrors.Errorf("DeleteBlock called but not implemented")
+}
+
+func (b *Blockstore) Put(block blocks.Block) error {
+	return xerrors.Errorf("Put called but not implemented")
+}
+
+func (b *Blockstore) PutMany(blocks []blocks.Block) error {
+	return xerrors.Errorf("PutMany called but not implemented")
+}

markets/dagstore/miner_api.go

@@ -4,8 +4,6 @@ import (
 	"context"
 	"fmt"
 	"io"
-	"os"
-	"strconv"
 
 	"github.com/filecoin-project/dagstore/throttle"
 	"github.com/ipfs/go-cid"
@@ -16,49 +14,84 @@ import (
 	"github.com/filecoin-project/go-fil-markets/shared"
 )
 
-// MaxConcurrentStorageCalls caps the amount of concurrent calls to the
-// storage, so that we don't spam it during heavy processes like bulk migration.
-var MaxConcurrentStorageCalls = func() int {
-	// TODO replace env with config.toml attribute.
-	v, ok := os.LookupEnv("LOTUS_DAGSTORE_MOUNT_CONCURRENCY")
-	if ok {
-		concurrency, err := strconv.Atoi(v)
-		if err == nil {
-			return concurrency
-		}
-	}
-	return 100
-}()
-
-type LotusAccessor interface {
+type MinerAPI interface {
 	FetchUnsealedPiece(ctx context.Context, pieceCid cid.Cid) (io.ReadCloser, error)
 	GetUnpaddedCARSize(ctx context.Context, pieceCid cid.Cid) (uint64, error)
+	IsUnsealed(ctx context.Context, pieceCid cid.Cid) (bool, error)
 	Start(ctx context.Context) error
 }
 
-type lotusAccessor struct {
+type minerAPI struct {
 	pieceStore piecestore.PieceStore
 	rm         retrievalmarket.RetrievalProviderNode
 	throttle   throttle.Throttler
 	readyMgr   *shared.ReadyManager
 }
 
-var _ LotusAccessor = (*lotusAccessor)(nil)
+var _ MinerAPI = (*minerAPI)(nil)
 
-func NewLotusAccessor(store piecestore.PieceStore, rm retrievalmarket.RetrievalProviderNode) LotusAccessor {
-	return &lotusAccessor{
+func NewMinerAPI(store piecestore.PieceStore, rm retrievalmarket.RetrievalProviderNode, concurrency int) MinerAPI {
+	return &minerAPI{
 		pieceStore: store,
 		rm:         rm,
-		throttle:   throttle.Fixed(MaxConcurrentStorageCalls),
+		throttle:   throttle.Fixed(concurrency),
 		readyMgr:   shared.NewReadyManager(),
 	}
 }
 
-func (m *lotusAccessor) Start(_ context.Context) error {
+func (m *minerAPI) Start(_ context.Context) error {
 	return m.readyMgr.FireReady(nil)
 }
 
-func (m *lotusAccessor) FetchUnsealedPiece(ctx context.Context, pieceCid cid.Cid) (io.ReadCloser, error) {
+func (m *minerAPI) IsUnsealed(ctx context.Context, pieceCid cid.Cid) (bool, error) {
+	err := m.readyMgr.AwaitReady()
+	if err != nil {
+		return false, xerrors.Errorf("failed while waiting for accessor to start: %w", err)
+	}
+
+	var pieceInfo piecestore.PieceInfo
+	err = m.throttle.Do(ctx, func(ctx context.Context) (err error) {
+		pieceInfo, err = m.pieceStore.GetPieceInfo(pieceCid)
+		return err
+	})
+
+	if err != nil {
+		return false, xerrors.Errorf("failed to fetch pieceInfo for piece %s: %w", pieceCid, err)
+	}
+
+	if len(pieceInfo.Deals) == 0 {
+		return false, xerrors.Errorf("no storage deals found for piece %s", pieceCid)
+	}
+
+	// check if we have an unsealed deal for the given piece in any of the unsealed sectors.
+	for _, deal := range pieceInfo.Deals {
+		deal := deal
+
+		var isUnsealed bool
+		// Throttle this path to avoid flooding the storage subsystem.
+		err := m.throttle.Do(ctx, func(ctx context.Context) (err error) {
+			isUnsealed, err = m.rm.IsUnsealed(ctx, deal.SectorID, deal.Offset.Unpadded(), deal.Length.Unpadded())
+			if err != nil {
+				return fmt.Errorf("failed to check if sector %d for deal %d was unsealed: %w", deal.SectorID, deal.DealID, err)
+			}
+			return nil
+		})
+
+		if err != nil {
+			log.Warnf("failed to check/retrieve unsealed sector: %s", err)
+			continue // move on to the next match.
+		}
+
+		if isUnsealed {
+			return true, nil
+		}
+	}
+
+	// we don't have an unsealed sector containing the piece
+	return false, nil
+}
+
+func (m *minerAPI) FetchUnsealedPiece(ctx context.Context, pieceCid cid.Cid) (io.ReadCloser, error) {
 	err := m.readyMgr.AwaitReady()
 	if err != nil {
 		return nil, err
@@ -130,7 +163,7 @@ func (m *lotusAccessor) FetchUnsealedPiece(ctx context.Context, pieceCid cid.Cid
 	return nil, lastErr
 }
 
-func (m *lotusAccessor) GetUnpaddedCARSize(ctx context.Context, pieceCid cid.Cid) (uint64, error) {
+func (m *minerAPI) GetUnpaddedCARSize(ctx context.Context, pieceCid cid.Cid) (uint64, error) {
 	err := m.readyMgr.AwaitReady()
 	if err != nil {
 		return 0, err

markets/dagstore/miner_api_test.go

@@ -45,7 +45,9 @@ func TestLotusAccessorFetchUnsealedPiece(t *testing.T) {
 		name        string
 		deals       []abi.SectorNumber
 		fetchedData string
-		expectErr   bool
+		isUnsealed  bool
+
+		expectErr bool
 	}{{
 		// Expect error if there is no deal info for piece CID
 		name:      "no deals",
@@ -56,11 +58,13 @@ func TestLotusAccessorFetchUnsealedPiece(t *testing.T) {
 		name:        "prefer unsealed deal",
 		deals:       []abi.SectorNumber{unsealedSectorID, sealedSectorID},
 		fetchedData: unsealedSectorData,
+		isUnsealed:  true,
 	}, {
 		// Expect the API to unseal the data if there are no unsealed deals
 		name:        "unseal if necessary",
 		deals:       []abi.SectorNumber{sealedSectorID},
 		fetchedData: sealedSectorData,
+		isUnsealed:  false,
 	}}
 
 	for _, tc := range testCases {
@@ -70,7 +74,7 @@ func TestLotusAccessorFetchUnsealedPiece(t *testing.T) {
 			rpn := &mockRPN{
 				sectors: mockData,
 			}
-			api := NewLotusAccessor(ps, rpn)
+			api := NewMinerAPI(ps, rpn, 100)
 			require.NoError(t, api.Start(ctx))
 
 			// Add deals to piece store
@@ -95,6 +99,10 @@ func TestLotusAccessorFetchUnsealedPiece(t *testing.T) {
 			require.NoError(t, err)
 
 			require.Equal(t, tc.fetchedData, string(bz))
+
+			uns, err := api.IsUnsealed(ctx, cid1)
+			require.NoError(t, err)
+			require.Equal(t, tc.isUnsealed, uns)
 		})
 	}
 }
@@ -106,7 +114,7 @@ func TestLotusAccessorGetUnpaddedCARSize(t *testing.T) {
 
 	ps := getPieceStore(t)
 	rpn := &mockRPN{}
-	api := NewLotusAccessor(ps, rpn)
+	api := NewMinerAPI(ps, rpn, 100)
 	require.NoError(t, api.Start(ctx))
 
 	// Add a deal with data Length 10
@@ -123,8 +131,6 @@ func TestLotusAccessorGetUnpaddedCARSize(t *testing.T) {
 }
 
 func TestThrottle(t *testing.T) {
-	MaxConcurrentStorageCalls = 3
-
 	ctx := context.Background()
 	cid1, err := cid.Parse("bafkqaaa")
 	require.NoError(t, err)
@@ -135,7 +141,7 @@ func TestThrottle(t *testing.T) {
 			unsealedSectorID: "foo",
 		},
 	}
-	api := NewLotusAccessor(ps, rpn)
+	api := NewMinerAPI(ps, rpn, 3)
 	require.NoError(t, api.Start(ctx))
 
 	// Add a deal with data Length 10
@@ -162,7 +168,7 @@ func TestThrottle(t *testing.T) {
 	}
 
 	time.Sleep(500 * time.Millisecond)
-	require.EqualValues(t, MaxConcurrentStorageCalls, atomic.LoadInt32(&rpn.calls)) // throttled
+	require.EqualValues(t, 3, atomic.LoadInt32(&rpn.calls)) // throttled
 
 	// allow to proceed.
 	rpn.lk.Unlock()
@@ -179,8 +185,14 @@ func getPieceStore(t *testing.T) piecestore.PieceStore {
 	ps, err := piecestoreimpl.NewPieceStore(ds_sync.MutexWrap(ds.NewMapDatastore()))
 	require.NoError(t, err)
 
+	ch := make(chan struct{}, 1)
+	ps.OnReady(func(_ error) {
+		ch <- struct{}{}
+	})
+
 	err = ps.Start(context.Background())
 	require.NoError(t, err)
+	<-ch
 	return ps
 }
 

markets/dagstore/mocks/mock_lotus_accessor.go

@@ -1,5 +1,5 @@
 // Code generated by MockGen. DO NOT EDIT.
-// Source: markets/dagstore/lotusaccessor.go
+// Source: lotusaccessor.go
 
 // Package mock_dagstore is a generated GoMock package.
 package mock_dagstore
@@ -31,10 +31,6 @@ func NewMockLotusAccessor(ctrl *gomock.Controller) *MockLotusAccessor {
 	return mock
 }
 
-func (mr *MockLotusAccessor) Start(_ context.Context) error {
-	return nil
-}
-
 // EXPECT returns an object that allows the caller to indicate expected use.
 func (m *MockLotusAccessor) EXPECT() *MockLotusAccessorMockRecorder {
 	return m.recorder
@@ -69,3 +65,32 @@ func (mr *MockLotusAccessorMockRecorder) GetUnpaddedCARSize(ctx, pieceCid interf
 	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetUnpaddedCARSize", reflect.TypeOf((*MockLotusAccessor)(nil).GetUnpaddedCARSize), ctx, pieceCid)
 }
+
+// IsUnsealed mocks base method.
+func (m *MockLotusAccessor) IsUnsealed(ctx context.Context, pieceCid cid.Cid) (bool, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "IsUnsealed", ctx, pieceCid)
+	ret0, _ := ret[0].(bool)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// IsUnsealed indicates an expected call of IsUnsealed.
+func (mr *MockLotusAccessorMockRecorder) IsUnsealed(ctx, pieceCid interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "IsUnsealed", reflect.TypeOf((*MockLotusAccessor)(nil).IsUnsealed), ctx, pieceCid)
+}
+
+// Start mocks base method.
+func (m *MockLotusAccessor) Start(ctx context.Context) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Start", ctx)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// Start indicates an expected call of Start.
+func (mr *MockLotusAccessorMockRecorder) Start(ctx interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Start", reflect.TypeOf((*MockLotusAccessor)(nil).Start), ctx)
+}

markets/dagstore/mount.go

@@ -15,28 +15,29 @@ const lotusScheme = "lotus"
 
 var _ mount.Mount = (*LotusMount)(nil)
 
-// LotusMount is the Lotus implementation of a Sharded DAG Store Mount.
-// A Filecoin Piece is treated as a Shard by this implementation.
-type LotusMount struct {
-	Api      LotusAccessor
-	PieceCid cid.Cid
-}
-
-// NewLotusMountTemplate is called when registering a mount with
-// the DAG store registry.
+// mountTemplate returns a templated LotusMount containing the supplied API.
+//
+// It is called when registering a mount type with the mount registry
+// of the DAG store. It is used to reinstantiate mounts after a restart.
 //
-// The DAG store registry receives an instance of the mount (a "template").
 // When the registry needs to deserialize a mount it clones the template then
 // calls Deserialize on the cloned instance, which will have a reference to the
 // lotus mount API supplied here.
-func NewLotusMountTemplate(api LotusAccessor) *LotusMount {
-	return &LotusMount{Api: api}
+func mountTemplate(api MinerAPI) *LotusMount {
+	return &LotusMount{API: api}
 }
 
-func NewLotusMount(pieceCid cid.Cid, api LotusAccessor) (*LotusMount, error) {
+// LotusMount is a DAGStore mount implementation that fetches deal data
+// from a PieceCID.
+type LotusMount struct {
+	API      MinerAPI
+	PieceCid cid.Cid
+}
+
+func NewLotusMount(pieceCid cid.Cid, api MinerAPI) (*LotusMount, error) {
 	return &LotusMount{
 		PieceCid: pieceCid,
-		Api:      api,
+		API:      api,
 	}, nil
 }
 
@@ -51,13 +52,12 @@ func (l *LotusMount) Deserialize(u *url.URL) error {
 	if err != nil {
 		return xerrors.Errorf("failed to parse PieceCid from host '%s': %w", u.Host, err)
 	}
-
 	l.PieceCid = pieceCid
 	return nil
 }
 
 func (l *LotusMount) Fetch(ctx context.Context) (mount.Reader, error) {
-	r, err := l.Api.FetchUnsealedPiece(ctx, l.PieceCid)
+	r, err := l.API.FetchUnsealedPiece(ctx, l.PieceCid)
 	if err != nil {
 		return nil, xerrors.Errorf("failed to fetch unsealed piece %s: %w", l.PieceCid, err)
 	}
@@ -78,15 +78,20 @@ func (l *LotusMount) Close() error {
 }
 
 func (l *LotusMount) Stat(ctx context.Context) (mount.Stat, error) {
-	size, err := l.Api.GetUnpaddedCARSize(ctx, l.PieceCid)
+	size, err := l.API.GetUnpaddedCARSize(ctx, l.PieceCid)
 	if err != nil {
 		return mount.Stat{}, xerrors.Errorf("failed to fetch piece size for piece %s: %w", l.PieceCid, err)
 	}
+	isUnsealed, err := l.API.IsUnsealed(ctx, l.PieceCid)
+	if err != nil {
+		return mount.Stat{}, xerrors.Errorf("failed to verify if we have the unsealed piece %s: %w", l.PieceCid, err)
+	}
 
 	// TODO Mark false when storage deal expires.
 	return mount.Stat{
 		Exists: true,
 		Size:   int64(size),
+		Ready:  isUnsealed,
 	}, nil
 }
 

markets/dagstore/mount_test.go

@@ -27,6 +27,9 @@ func TestLotusMount(t *testing.T) {
 
 	// create a mock lotus api that returns the reader we want
 	mockLotusMountAPI := mock_dagstore.NewMockLotusAccessor(mockCtrl)
+
+	mockLotusMountAPI.EXPECT().IsUnsealed(gomock.Any(), cid).Return(true, nil).Times(1)
+
 	mockLotusMountAPI.EXPECT().FetchUnsealedPiece(gomock.Any(), cid).Return(&readCloser{ioutil.NopCloser(strings.NewReader("testing"))}, nil).Times(1)
 	mockLotusMountAPI.EXPECT().FetchUnsealedPiece(gomock.Any(), cid).Return(&readCloser{ioutil.NopCloser(strings.NewReader("testing"))}, nil).Times(1)
 	mockLotusMountAPI.EXPECT().GetUnpaddedCARSize(ctx, cid).Return(uint64(100), nil).Times(1)
@@ -52,7 +55,7 @@ func TestLotusMount(t *testing.T) {
 	// serialize url then deserialize from mount template -> should get back
 	// the same mount
 	url := mnt.Serialize()
-	mnt2 := NewLotusMountTemplate(mockLotusMountAPI)
+	mnt2 := mountTemplate(mockLotusMountAPI)
 	err = mnt2.Deserialize(url)
 	require.NoError(t, err)
 
@@ -66,7 +69,7 @@ func TestLotusMount(t *testing.T) {
 }
 
 func TestLotusMountDeserialize(t *testing.T) {
-	api := &lotusAccessor{}
+	api := &minerAPI{}
 
 	bgen := blocksutil.NewBlockGenerator()
 	cid := bgen.Next().Cid()
@@ -76,12 +79,12 @@ func TestLotusMountDeserialize(t *testing.T) {
 	u, err := url.Parse(us)
 	require.NoError(t, err)
 
-	mnt := NewLotusMountTemplate(api)
+	mnt := mountTemplate(api)
 	err = mnt.Deserialize(u)
 	require.NoError(t, err)
 
 	require.Equal(t, cid, mnt.PieceCid)
-	require.Equal(t, api, mnt.Api)
+	require.Equal(t, api, mnt.API)
 
 	// fails if cid is not valid
 	us = lotusScheme + "://" + "rand"
@@ -108,14 +111,16 @@ func TestLotusMountRegistration(t *testing.T) {
 
 	mockLotusMountAPI := mock_dagstore.NewMockLotusAccessor(mockCtrl)
 	registry := mount.NewRegistry()
-	err = registry.Register(lotusScheme, NewLotusMountTemplate(mockLotusMountAPI))
+	err = registry.Register(lotusScheme, mountTemplate(mockLotusMountAPI))
 	require.NoError(t, err)
 
 	mnt, err := registry.Instantiate(u)
 	require.NoError(t, err)
 
+	mockLotusMountAPI.EXPECT().IsUnsealed(ctx, cid).Return(true, nil)
 	mockLotusMountAPI.EXPECT().GetUnpaddedCARSize(ctx, cid).Return(uint64(100), nil).Times(1)
 	stat, err := mnt.Stat(context.Background())
 	require.NoError(t, err)
 	require.EqualValues(t, 100, stat.Size)
+	require.True(t, stat.Ready)
 }

markets/dagstore/readonlyblockstore.go

@@ -1,33 +0,0 @@
-package dagstore
-
-import (
-	blocks "github.com/ipfs/go-block-format"
-	"github.com/ipfs/go-cid"
-	bstore "github.com/ipfs/go-ipfs-blockstore"
-	"golang.org/x/xerrors"
-
-	"github.com/filecoin-project/dagstore"
-)
-
-// ReadOnlyBlockstore stubs out Blockstore mutators with methods that error out
-type ReadOnlyBlockstore struct {
-	dagstore.ReadBlockstore
-}
-
-func NewReadOnlyBlockstore(rbs dagstore.ReadBlockstore) bstore.Blockstore {
-	return ReadOnlyBlockstore{ReadBlockstore: rbs}
-}
-
-func (r ReadOnlyBlockstore) DeleteBlock(c cid.Cid) error {
-	return xerrors.Errorf("DeleteBlock called but not implemented")
-}
-
-func (r ReadOnlyBlockstore) Put(block blocks.Block) error {
-	return xerrors.Errorf("Put called but not implemented")
-}
-
-func (r ReadOnlyBlockstore) PutMany(blocks []blocks.Block) error {
-	return xerrors.Errorf("PutMany called but not implemented")
-}
-
-var _ bstore.Blockstore = (*ReadOnlyBlockstore)(nil)

markets/dagstore/wrapper.go

@@ -3,174 +3,202 @@ package dagstore
 import (
 	"context"
 	"errors"
-	"io"
+	"math"
+	"os"
+	"path/filepath"
 	"sync"
 	"time"
 
-	"github.com/filecoin-project/dagstore/index"
 	"github.com/ipfs/go-cid"
 	ds "github.com/ipfs/go-datastore"
-	bstore "github.com/ipfs/go-ipfs-blockstore"
+	levelds "github.com/ipfs/go-ds-leveldb"
+	measure "github.com/ipfs/go-ds-measure"
 	logging "github.com/ipfs/go-log/v2"
+	ldbopts "github.com/syndtr/goleveldb/leveldb/opt"
 	"golang.org/x/xerrors"
 
+	"github.com/filecoin-project/lotus/node/config"
+
+	"github.com/filecoin-project/go-statemachine/fsm"
+
 	"github.com/filecoin-project/dagstore"
+	"github.com/filecoin-project/dagstore/index"
 	"github.com/filecoin-project/dagstore/mount"
 	"github.com/filecoin-project/dagstore/shard"
-	"github.com/filecoin-project/go-fil-markets/carstore"
-	"github.com/filecoin-project/go-fil-markets/shared"
+
+	"github.com/filecoin-project/go-fil-markets/storagemarket"
+	"github.com/filecoin-project/go-fil-markets/storagemarket/impl/providerstates"
+	"github.com/filecoin-project/go-fil-markets/stores"
 )
 
-const maxRecoverAttempts = 1
+const (
+	maxRecoverAttempts = 1
+	shardRegMarker     = ".shard-registration-complete"
+)
 
-var log = logging.Logger("dagstore-wrapper")
-
-// MarketDAGStoreConfig is the config the market needs to then construct a DAG Store.
-type MarketDAGStoreConfig struct {
-	TransientsDir       string
-	IndexDir            string
-	Datastore           ds.Datastore
-	MaxConcurrentIndex  int
-	MaxConcurrentCopies int
-	GCInterval          time.Duration
-}
-
-// DAGStore provides an interface for the DAG store that can be mocked out
-// by tests
-type DAGStore interface {
-	RegisterShard(ctx context.Context, key shard.Key, mnt mount.Mount, out chan dagstore.ShardResult, opts dagstore.RegisterOpts) error
-	AcquireShard(ctx context.Context, key shard.Key, out chan dagstore.ShardResult, _ dagstore.AcquireOpts) error
-	RecoverShard(ctx context.Context, key shard.Key, out chan dagstore.ShardResult, _ dagstore.RecoverOpts) error
-	GC(ctx context.Context) (*dagstore.GCResult, error)
-	Close() error
-	Start(ctx context.Context) error
-}
-
-type closableBlockstore struct {
-	bstore.Blockstore
-	io.Closer
-}
+var log = logging.Logger("dagstore")
 
 type Wrapper struct {
 	ctx          context.Context
 	cancel       context.CancelFunc
 	backgroundWg sync.WaitGroup
 
-	dagStore   DAGStore
-	mountApi   LotusAccessor
+	cfg        config.DAGStoreConfig
+	dagst      dagstore.Interface
+	minerAPI   MinerAPI
 	failureCh  chan dagstore.ShardResult
 	traceCh    chan dagstore.Trace
 	gcInterval time.Duration
 }
 
-var _ shared.DagStoreWrapper = (*Wrapper)(nil)
+var _ stores.DAGStoreWrapper = (*Wrapper)(nil)
 
-func NewDagStoreWrapper(cfg MarketDAGStoreConfig, mountApi LotusAccessor) (*Wrapper, error) {
+func NewDAGStore(cfg config.DAGStoreConfig, mountApi MinerAPI) (*dagstore.DAGStore, *Wrapper, error) {
 	// construct the DAG Store.
 	registry := mount.NewRegistry()
-	if err := registry.Register(lotusScheme, NewLotusMountTemplate(mountApi)); err != nil {
-		return nil, xerrors.Errorf("failed to create registry: %w", err)
+	if err := registry.Register(lotusScheme, mountTemplate(mountApi)); err != nil {
+		return nil, nil, xerrors.Errorf("failed to create registry: %w", err)
 	}
 
 	// The dagstore will write Shard failures to the `failureCh` here.
 	failureCh := make(chan dagstore.ShardResult, 1)
+
 	// The dagstore will write Trace events to the `traceCh` here.
 	traceCh := make(chan dagstore.Trace, 32)
 
-	irepo, err := index.NewFSRepo(cfg.IndexDir)
+	var (
+		transientsDir = filepath.Join(cfg.RootDir, "transients")
+		datastoreDir  = filepath.Join(cfg.RootDir, "datastore")
+		indexDir      = filepath.Join(cfg.RootDir, "index")
+	)
+
+	dstore, err := newDatastore(datastoreDir)
 	if err != nil {
-		return nil, xerrors.Errorf("failed to initialise dagstore index repo")
+		return nil, nil, xerrors.Errorf("failed to create dagstore datastore in %s: %w", datastoreDir, err)
+	}
+
+	irepo, err := index.NewFSRepo(indexDir)
+	if err != nil {
+		return nil, nil, xerrors.Errorf("failed to initialise dagstore index repo")
 	}
 
 	dcfg := dagstore.Config{
-		TransientsDir: cfg.TransientsDir,
+		TransientsDir: transientsDir,
 		IndexRepo:     irepo,
-		Datastore:     cfg.Datastore,
+		Datastore:     dstore,
 		MountRegistry: registry,
 		FailureCh:     failureCh,
 		TraceCh:       traceCh,
 		// not limiting fetches globally, as the Lotus mount does
 		// conditional throttling.
-		MaxConcurrentIndex:  cfg.MaxConcurrentIndex,
-		MaxConcurrentCopies: cfg.MaxConcurrentCopies,
-		RecoverOnStart:      dagstore.RecoverOnAcquire,
-	}
-	dagStore, err := dagstore.NewDAGStore(dcfg)
-	if err != nil {
-		return nil, xerrors.Errorf("failed to create DAG store: %w", err)
+		MaxConcurrentIndex:        cfg.MaxConcurrentIndex,
+		MaxConcurrentReadyFetches: cfg.MaxConcurrentReadyFetches,
+		RecoverOnStart:            dagstore.RecoverOnAcquire,
 	}
 
-	return &Wrapper{
-		dagStore:   dagStore,
-		mountApi:   mountApi,
+	dagst, err := dagstore.NewDAGStore(dcfg)
+	if err != nil {
+		return nil, nil, xerrors.Errorf("failed to create DAG store: %w", err)
+	}
+
+	w := &Wrapper{
+		cfg:        cfg,
+		dagst:      dagst,
+		minerAPI:   mountApi,
 		failureCh:  failureCh,
 		traceCh:    traceCh,
-		gcInterval: cfg.GCInterval,
-	}, nil
-}
-
-func (ds *Wrapper) Start(ctx context.Context) error {
-	ds.ctx, ds.cancel = context.WithCancel(ctx)
-
-	// Run a go-routine to do DagStore GC.
-	ds.backgroundWg.Add(1)
-	go ds.dagStoreGCLoop()
-
-	// run a go-routine to read the trace for debugging.
-	ds.backgroundWg.Add(1)
-	go ds.traceLoop()
-
-	// Run a go-routine for shard recovery
-	if dss, ok := ds.dagStore.(*dagstore.DAGStore); ok {
-		ds.backgroundWg.Add(1)
-		go dagstore.RecoverImmediately(ds.ctx, dss, ds.failureCh, maxRecoverAttempts, ds.backgroundWg.Done)
+		gcInterval: time.Duration(cfg.GCInterval),
 	}
 
-	return ds.dagStore.Start(ctx)
+	return dagst, w, nil
 }
 
-func (ds *Wrapper) traceLoop() {
-	defer ds.backgroundWg.Done()
+// newDatastore creates a datastore under the given base directory
+// for dagstore metadata.
+func newDatastore(dir string) (ds.Batching, error) {
+	// Create the datastore directory if it doesn't exist yet.
+	if err := os.MkdirAll(dir, 0755); err != nil {
+		return nil, xerrors.Errorf("failed to create directory %s for DAG store datastore: %w", dir, err)
+	}
 
-	for ds.ctx.Err() == nil {
+	// Create a new LevelDB datastore
+	dstore, err := levelds.NewDatastore(dir, &levelds.Options{
+		Compression: ldbopts.NoCompression,
+		NoSync:      false,
+		Strict:      ldbopts.StrictAll,
+		ReadOnly:    false,
+	})
+	if err != nil {
+		return nil, xerrors.Errorf("failed to open datastore for DAG store: %w", err)
+	}
+	// Keep statistics about the datastore
+	mds := measure.New("measure.", dstore)
+	return mds, nil
+}
+
+func (w *Wrapper) Start(ctx context.Context) error {
+	w.ctx, w.cancel = context.WithCancel(ctx)
+
+	// Run a go-routine to do DagStore GC.
+	w.backgroundWg.Add(1)
+	go w.gcLoop()
+
+	// run a go-routine to read the trace for debugging.
+	w.backgroundWg.Add(1)
+	go w.traceLoop()
+
+	// Run a go-routine for shard recovery
+	if dss, ok := w.dagst.(*dagstore.DAGStore); ok {
+		w.backgroundWg.Add(1)
+		go dagstore.RecoverImmediately(w.ctx, dss, w.failureCh, maxRecoverAttempts, w.backgroundWg.Done)
+	}
+
+	return w.dagst.Start(ctx)
+}
+
+func (w *Wrapper) traceLoop() {
+	defer w.backgroundWg.Done()
+
+	for w.ctx.Err() == nil {
 		select {
 		// Log trace events from the DAG store
-		case tr := <-ds.traceCh:
+		case tr := <-w.traceCh:
 			log.Debugw("trace",
 				"shard-key", tr.Key.String(),
 				"op-type", tr.Op.String(),
 				"after", tr.After.String())
 
-		case <-ds.ctx.Done():
+		case <-w.ctx.Done():
 			return
 		}
 	}
 }
 
-func (ds *Wrapper) dagStoreGCLoop() {
-	defer ds.backgroundWg.Done()
+func (w *Wrapper) gcLoop() {
+	defer w.backgroundWg.Done()
 
-	gcTicker := time.NewTicker(ds.gcInterval)
-	defer gcTicker.Stop()
+	ticker := time.NewTicker(w.gcInterval)
+	defer ticker.Stop()
 
-	for ds.ctx.Err() == nil {
+	for w.ctx.Err() == nil {
 		select {
 		// GC the DAG store on every tick
-		case <-gcTicker.C:
-			_, _ = ds.dagStore.GC(ds.ctx)
+		case <-ticker.C:
+			_, _ = w.dagst.GC(w.ctx)
 
 		// Exit when the DAG store wrapper is shutdown
-		case <-ds.ctx.Done():
+		case <-w.ctx.Done():
 			return
 		}
 	}
 }
 
-func (ds *Wrapper) LoadShard(ctx context.Context, pieceCid cid.Cid) (carstore.ClosableBlockstore, error) {
+func (w *Wrapper) LoadShard(ctx context.Context, pieceCid cid.Cid) (stores.ClosableBlockstore, error) {
 	log.Debugf("acquiring shard for piece CID %s", pieceCid)
+
 	key := shard.KeyFromCID(pieceCid)
 	resch := make(chan dagstore.ShardResult, 1)
-	err := ds.dagStore.AcquireShard(ctx, key, resch, dagstore.AcquireOpts{})
+	err := w.dagst.AcquireShard(ctx, key, resch, dagstore.AcquireOpts{})
 	log.Debugf("sent message to acquire shard for piece CID %s", pieceCid)
 
 	if err != nil {
@@ -185,13 +213,13 @@ func (ds *Wrapper) LoadShard(ctx context.Context, pieceCid cid.Cid) (carstore.Cl
 		// we already have a transient file. However, we don't have it here
 		// and therefore we pass an empty file path.
 		carPath := ""
-		if err := shared.RegisterShardSync(ctx, ds, pieceCid, carPath, false); err != nil {
+		if err := stores.RegisterShardSync(ctx, w, pieceCid, carPath, false); err != nil {
 			return nil, xerrors.Errorf("failed to re-register shard during loading piece CID %s: %w", pieceCid, err)
 		}
 		log.Warnw("successfully re-registered shard", "pieceCID", pieceCid)
 
 		resch = make(chan dagstore.ShardResult, 1)
-		if err := ds.dagStore.AcquireShard(ctx, key, resch, dagstore.AcquireOpts{}); err != nil {
+		if err := w.dagst.AcquireShard(ctx, key, resch, dagstore.AcquireOpts{}); err != nil {
 			return nil, xerrors.Errorf("failed to acquire Shard for piece CID %s after re-registering: %w", pieceCid, err)
 		}
 	}
@@ -216,13 +244,13 @@ func (ds *Wrapper) LoadShard(ctx context.Context, pieceCid cid.Cid) (carstore.Cl
 	}
 
 	log.Debugf("successfully loaded blockstore for piece CID %s", pieceCid)
-	return &closableBlockstore{Blockstore: NewReadOnlyBlockstore(bs), Closer: res.Accessor}, nil
+	return &Blockstore{ReadBlockstore: bs, Closer: res.Accessor}, nil
 }
 
-func (ds *Wrapper) RegisterShard(ctx context.Context, pieceCid cid.Cid, carPath string, eagerInit bool, resch chan dagstore.ShardResult) error {
+func (w *Wrapper) RegisterShard(ctx context.Context, pieceCid cid.Cid, carPath string, eagerInit bool, resch chan dagstore.ShardResult) error {
 	// Create a lotus mount with the piece CID
 	key := shard.KeyFromCID(pieceCid)
-	mt, err := NewLotusMount(pieceCid, ds.mountApi)
+	mt, err := NewLotusMount(pieceCid, w.minerAPI)
 	if err != nil {
 		return xerrors.Errorf("failed to create lotus mount for piece CID %s: %w", pieceCid, err)
 	}
@@ -232,7 +260,7 @@ func (ds *Wrapper) RegisterShard(ctx context.Context, pieceCid cid.Cid, carPath
 		ExistingTransient:  carPath,
 		LazyInitialization: !eagerInit,
 	}
-	err = ds.dagStore.RegisterShard(ctx, key, mt, resch, opts)
+	err = w.dagst.RegisterShard(ctx, key, mt, resch, opts)
 	if err != nil {
 		return xerrors.Errorf("failed to schedule register shard for piece CID %s: %w", pieceCid, err)
 	}
@@ -241,21 +269,153 @@ func (ds *Wrapper) RegisterShard(ctx context.Context, pieceCid cid.Cid, carPath
 	return nil
 }
 
-func (ds *Wrapper) Close() error {
+func (w *Wrapper) MigrateDeals(ctx context.Context, deals []storagemarket.MinerDeal) (bool, error) {
+	log := log.Named("migrator")
+
+	// Check if all deals have already been registered as shards
+	isComplete, err := w.registrationComplete()
+	if err != nil {
+		return false, xerrors.Errorf("failed to get dagstore migration status: %w", err)
+	}
+	if isComplete {
+		// All deals have been registered as shards, bail out
+		log.Info("no shard migration necessary; already marked complete")
+		return false, nil
+	}
+
+	log.Infow("registering shards for all active deals in sealing subsystem", "count", len(deals))
+
+	inSealingSubsystem := make(map[fsm.StateKey]struct{}, len(providerstates.StatesKnownBySealingSubsystem))
+	for _, s := range providerstates.StatesKnownBySealingSubsystem {
+		inSealingSubsystem[s] = struct{}{}
+	}
+
+	// channel where results will be received, and channel where the total
+	// number of registered shards will be sent.
+	resch := make(chan dagstore.ShardResult, 32)
+	totalCh := make(chan int)
+	doneCh := make(chan struct{})
+
+	// Start making progress consuming results. We won't know how many to
+	// actually consume until we register all shards.
+	//
+	// If there are any problems registering shards, just log an error
+	go func() {
+		defer close(doneCh)
+
+		var total = math.MaxInt64
+		var res dagstore.ShardResult
+		for rcvd := 0; rcvd < total; {
+			select {
+			case total = <-totalCh:
+				// we now know the total number of registered shards
+				// nullify so that we no longer consume from it after closed.
+				close(totalCh)
+				totalCh = nil
+			case res = <-resch:
+				rcvd++
+				if res.Error == nil {
+					log.Infow("async shard registration completed successfully", "shard_key", res.Key)
+				} else {
+					log.Warnw("async shard registration failed", "shard_key", res.Key, "error", res.Error)
+				}
+			}
+		}
+	}()
+
+	// Filter for deals that are handed off.
+	//
+	// If the deal has not yet been handed off to the sealing subsystem, we
+	// don't need to call RegisterShard in this migration; RegisterShard will
+	// be called in the new code once the deal reaches the state where it's
+	// handed off to the sealing subsystem.
+	var registered int
+	for _, deal := range deals {
+		if deal.Ref.PieceCid == nil {
+			log.Warnw("deal has nil piece CID; skipping", "deal_id", deal.DealID)
+			continue
+		}
+
+		// enrich log statements in this iteration with deal ID and piece CID.
+		log := log.With("deal_id", deal.DealID, "piece_cid", deal.Ref.PieceCid)
+
+		// Filter for deals that have been handed off to the sealing subsystem
+		if _, ok := inSealingSubsystem[deal.State]; !ok {
+			log.Infow("deal not ready; skipping")
+			continue
+		}
+
+		log.Infow("registering deal in dagstore with lazy init")
+
+		// Register the deal as a shard with the DAG store with lazy initialization.
+		// The index will be populated the first time the deal is retrieved, or
+		// through the bulk initialization script.
+		err = w.RegisterShard(ctx, *deal.Ref.PieceCid, "", false, resch)
+		if err != nil {
+			log.Warnw("failed to register shard", "error", err)
+			continue
+		}
+		registered++
+	}
+
+	log.Infow("finished registering all shards", "total", registered)
+	totalCh <- registered
+	<-doneCh
+
+	log.Infow("confirmed registration of all shards")
+
+	// Completed registering all shards, so mark the migration as complete
+	err = w.markRegistrationComplete()
+	if err != nil {
+		log.Errorf("failed to mark shards as registered: %s", err)
+	} else {
+		log.Info("successfully marked migration as complete")
+	}
+
+	log.Infow("dagstore migration complete")
+
+	return true, nil
+}
+
+// Check for the existence of a "marker" file indicating that the migration
+// has completed
+func (w *Wrapper) registrationComplete() (bool, error) {
+	path := filepath.Join(w.cfg.RootDir, shardRegMarker)
+	_, err := os.Stat(path)
+	if os.IsNotExist(err) {
+		return false, nil
+	}
+	if err != nil {
+		return false, err
+	}
+	return true, nil
+}
+
+// Create a "marker" file indicating that the migration has completed
+func (w *Wrapper) markRegistrationComplete() error {
+	path := filepath.Join(w.cfg.RootDir, shardRegMarker)
+	file, err := os.Create(path)
+	if err != nil {
+		return err
+	}
+	return file.Close()
+}
+
+func (w *Wrapper) Close() error {
 	// Cancel the context
-	ds.cancel()
+	w.cancel()
 
 	// Close the DAG store
 	log.Info("will close the dagstore")
-	if err := ds.dagStore.Close(); err != nil {
-		return xerrors.Errorf("failed to close DAG store: %w", err)
+	if err := w.dagst.Close(); err != nil {
+		return xerrors.Errorf("failed to close dagstore: %w", err)
 	}
 	log.Info("dagstore closed")
 
 	// Wait for the background go routine to exit
-	log.Info("waiting for dagstore background wrapper routines to exist")
-	ds.backgroundWg.Wait()
-	log.Info("exited dagstore background warpper routines")
+	log.Info("waiting for dagstore background wrapper goroutines to exit")
+	w.backgroundWg.Wait()
+	log.Info("exited dagstore background wrapper goroutines")
 
 	return nil
 }

markets/dagstore/wrapper_migration_test.go

@@ -0,0 +1,110 @@
+package dagstore
+
+import (
+	"context"
+	"testing"
+
+	"github.com/filecoin-project/dagstore"
+	"github.com/stretchr/testify/require"
+
+	"github.com/filecoin-project/go-state-types/abi"
+
+	"github.com/filecoin-project/go-fil-markets/retrievalmarket/impl/testnodes"
+	tut "github.com/filecoin-project/go-fil-markets/shared_testutil"
+	"github.com/filecoin-project/go-fil-markets/storagemarket"
+	"github.com/filecoin-project/lotus/node/config"
+)
+
+func TestShardRegistration(t *testing.T) {
+	ps := tut.NewTestPieceStore()
+	providerNode := testnodes.NewTestRetrievalProviderNode()
+
+	ctx := context.Background()
+	cids := tut.GenerateCids(4)
+	pieceCidUnsealed := cids[0]
+	pieceCidSealed := cids[1]
+	pieceCidUnsealed2 := cids[2]
+	pieceCidUnsealed3 := cids[3]
+
+	sealedSector := abi.SectorNumber(1)
+	unsealedSector1 := abi.SectorNumber(2)
+	unsealedSector2 := abi.SectorNumber(3)
+	unsealedSector3 := abi.SectorNumber(4)
+
+	// ps.ExpectPiece(pieceCidUnsealed, piecestore.PieceInfo{
+	// 	PieceCID: pieceCidUnsealed,
+	// 	Deals: []piecestore.DealInfo{
+	// 		{
+	// 			SectorID: unsealedSector1,
+	// 		},
+	// 	},
+	// })
+	//
+	// ps.ExpectPiece(pieceCidSealed, piecestore.PieceInfo{
+	// 	PieceCID: pieceCidSealed,
+	// 	Deals: []piecestore.DealInfo{
+	// 		{
+	// 			SectorID: sealedSector,
+	// 		},
+	// 	},
+	// })
+
+	deals := []storagemarket.MinerDeal{{
+		// Should be registered
+		State:        storagemarket.StorageDealSealing,
+		SectorNumber: unsealedSector1,
+		Ref: &storagemarket.DataRef{
+			PieceCid: &pieceCidUnsealed,
+		},
+	}, {
+		// Should be registered with lazy registration (because sector is sealed)
+		State:        storagemarket.StorageDealSealing,
+		SectorNumber: sealedSector,
+		Ref: &storagemarket.DataRef{
+			PieceCid: &pieceCidSealed,
+		},
+	}, {
+		// Should be ignored because deal is no longer active
+		State:        storagemarket.StorageDealError,
+		SectorNumber: unsealedSector2,
+		Ref: &storagemarket.DataRef{
+			PieceCid: &pieceCidUnsealed2,
+		},
+	}, {
+		// Should be ignored because deal is not yet sealing
+		State:        storagemarket.StorageDealFundsReserved,
+		SectorNumber: unsealedSector3,
+		Ref: &storagemarket.DataRef{
+			PieceCid: &pieceCidUnsealed3,
+		},
+	}}
+
+	cfg := config.DefaultStorageMiner().DAGStore
+	cfg.RootDir = t.TempDir()
+
+	mapi := NewMinerAPI(ps, providerNode, 10)
+	dagst, w, err := NewDAGStore(cfg, mapi)
+	require.NoError(t, err)
+	require.NotNil(t, dagst)
+	require.NotNil(t, w)
+
+	err = dagst.Start(context.Background())
+	require.NoError(t, err)
+
+	migrated, err := w.MigrateDeals(ctx, deals)
+	require.True(t, migrated)
+	require.NoError(t, err)
+
+	info := dagst.AllShardsInfo()
+	require.Len(t, info, 2)
+	for _, i := range info {
+		require.Equal(t, dagstore.ShardStateNew, i.ShardState)
+	}
+
+	// Run register shard migration again
+	migrated, err = w.MigrateDeals(ctx, deals)
+	require.False(t, migrated)
+	require.NoError(t, err)
+
+	// ps.VerifyExpectations(t)
+}

markets/dagstore/wrapper_test.go

@@ -10,6 +10,8 @@ import (
 
 	"golang.org/x/xerrors"
 
+	"github.com/filecoin-project/lotus/node/config"
+
 	"github.com/filecoin-project/dagstore"
 	"github.com/filecoin-project/dagstore/mount"
 	"github.com/filecoin-project/dagstore/shard"
@@ -26,13 +28,14 @@ func TestWrapperAcquireRecovery(t *testing.T) {
 	require.NoError(t, err)
 
 	// Create a DAG store wrapper
-	w, err := NewDagStoreWrapper(MarketDAGStoreConfig{
-		TransientsDir: t.TempDir(),
-		IndexDir:      t.TempDir(),
-		GCInterval:    time.Millisecond,
+	dagst, w, err := NewDAGStore(config.DAGStoreConfig{
+		RootDir:    t.TempDir(),
+		GCInterval: config.Duration(1 * time.Millisecond),
 	}, mockLotusMount{})
 	require.NoError(t, err)
 
+	defer dagst.Close() //nolint:errcheck
+
 	// Return an error from acquire shard the first time
 	acquireShardErr := make(chan error, 1)
 	acquireShardErr <- xerrors.Errorf("unknown shard: %w", dagstore.ErrShardUnknown)
@@ -45,7 +48,7 @@ func TestWrapperAcquireRecovery(t *testing.T) {
 		},
 		register: make(chan shard.Key, 1),
 	}
-	w.dagStore = mock
+	w.dagst = mock
 
 	mybs, err := w.LoadShard(ctx, pieceCid)
 	require.NoError(t, err)
@@ -76,23 +79,25 @@ func TestWrapperBackground(t *testing.T) {
 	ctx := context.Background()
 
 	// Create a DAG store wrapper
-	w, err := NewDagStoreWrapper(MarketDAGStoreConfig{
-		TransientsDir: t.TempDir(),
-		IndexDir:      t.TempDir(),
-		GCInterval:    time.Millisecond,
+	dagst, w, err := NewDAGStore(config.DAGStoreConfig{
+		RootDir:    t.TempDir(),
+		GCInterval: config.Duration(1 * time.Millisecond),
 	}, mockLotusMount{})
 	require.NoError(t, err)
 
+	defer dagst.Close() //nolint:errcheck
+
 	// Create a mock DAG store in place of the real DAG store
 	mock := &mockDagStore{
 		gc:      make(chan struct{}, 1),
 		recover: make(chan shard.Key, 1),
 		close:   make(chan struct{}, 1),
 	}
-	w.dagStore = mock
+	w.dagst = mock
 
 	// Start up the wrapper
-	w.Start(ctx)
+	err = w.Start(ctx)
+	require.NoError(t, err)
 
 	// Expect GC to be called automatically
 	tctx, cancel := context.WithTimeout(ctx, time.Second)
@@ -127,6 +132,22 @@ type mockDagStore struct {
 	close   chan struct{}
 }
 
+func (m *mockDagStore) DestroyShard(ctx context.Context, key shard.Key, out chan dagstore.ShardResult, _ dagstore.DestroyOpts) error {
+	panic("implement me")
+}
+
+func (m *mockDagStore) GetShardInfo(k shard.Key) (dagstore.ShardInfo, error) {
+	panic("implement me")
+}
+
+func (m *mockDagStore) AllShardsInfo() dagstore.AllShardsInfo {
+	panic("implement me")
+}
+
+func (m *mockDagStore) Start(_ context.Context) error {
+	return nil
+}
+
 func (m *mockDagStore) RegisterShard(ctx context.Context, key shard.Key, mnt mount.Mount, out chan dagstore.ShardResult, opts dagstore.RegisterOpts) error {
 	m.register <- key
 	out <- dagstore.ShardResult{Key: key}
@@ -178,6 +199,10 @@ func (m mockLotusMount) GetUnpaddedCARSize(ctx context.Context, pieceCid cid.Cid
 	panic("implement me")
 }
 
+func (m mockLotusMount) IsUnsealed(ctx context.Context, pieceCid cid.Cid) (bool, error) {
+	panic("implement me")
+}
+
 func getShardAccessor(t *testing.T) *dagstore.ShardAccessor {
 	data, err := os.ReadFile("./fixtures/sample-rw-bs-v2.car")
 	require.NoError(t, err)

node/builder.go

@@ -66,6 +66,7 @@ var (
 	AutoNATSvcKey        = special{10} // Libp2p option
 	BandwidthReporterKey = special{11} // Libp2p option
 	ConnGaterKey         = special{12} // libp2p option
+	DAGStoreKey          = special{13} // constructor returns multiple values
 )
 
 type invoke int

node/builder_miner.go

@@ -148,8 +148,8 @@ func ConfigStorageMiner(c interface{}) Option {
 			Override(new(dtypes.RetrievalPricingFunc), modules.RetrievalPricingFunc(cfg.Dealmaking)),
 
 			// DAG Store
-			Override(new(dagstore.LotusAccessor), modules.NewLotusAccessor),
-			Override(new(*dagstore.Wrapper), modules.DagStoreWrapper),
+			Override(new(dagstore.MinerAPI), modules.NewMinerAPI),
+			Override(DAGStoreKey, modules.DAGStore),
 
 			// Markets (retrieval)
 			Override(new(retrievalmarket.RetrievalProviderNode), retrievaladapter.NewRetrievalProviderNode),

node/config/def.go

@@ -188,6 +188,12 @@ func DefaultStorageMiner() *StorageMiner {
 			TerminateControl:   []string{},
 			DealPublishControl: []string{},
 		},
+
+		DAGStore: DAGStoreConfig{
+			MaxConcurrentIndex:         5,
+			MaxConcurrencyStorageCalls: 100,
+			GCInterval:                 Duration(1 * time.Minute),
+		},
 	}
 	cfg.Common.API.ListenAddress = "/ip4/127.0.0.1/tcp/2345/http"
 	cfg.Common.API.RemoteListenAddress = "127.0.0.1:2345"

node/config/doc_gen.go

@@ -125,6 +125,46 @@ and storage providers`,
 			Comment: ``,
 		},
 	},
+	"DAGStoreConfig": []DocField{
+		{
+			Name: "RootDir",
+			Type: "string",
+
+			Comment: `Default value: $LOTUS_MARKETS_PATH/dagStore (split deployment) or
+$LOTUS_MINER_PATH/dagStore (monolith deployment)`,
+		},
+		{
+			Name: "MaxConcurrentIndex",
+			Type: "int",
+
+			Comment: `The maximum amount of indexing jobs that can run simultaneously.
+Default value: 5.`,
+		},
+		{
+			Name: "MaxConcurrentReadyFetches",
+			Type: "int",
+
+			Comment: `The maximum amount of unsealed deals that can be fetched simultaneously
+from the storage subsystem.
+Default value: 2.`,
+		},
+		{
+			Name: "MaxConcurrencyStorageCalls",
+			Type: "int",
+
+			Comment: `The maximum number of simultaneous inflight API calls to the storage
+subsystem.
+Default value: 100.`,
+		},
+		{
+			Name: "GCInterval",
+			Type: "Duration",
+
+			Comment: `The time between calls to periodic dagstore GC, in time.Duration string
+representation, e.g. 1m, 5m, 1h.
+Default value: 1 minute.`,
+		},
+	},
 	"DealmakingConfig": []DocField{
 		{
 			Name: "ConsiderOnlineStorageDeals",
@@ -753,6 +793,12 @@ Default is 20 (about once a week).`,
 			Name: "Addresses",
 			Type: "MinerAddressConfig",
 
+			Comment: ``,
+		},
+		{
+			Name: "DAGStore",
+			Type: "DAGStoreConfig",
+
 			Comment: ``,
 		},
 	},

node/config/types.go

@@ -50,6 +50,42 @@ type StorageMiner struct {
 	Storage    sectorstorage.SealerConfig
 	Fees       MinerFeeConfig
 	Addresses  MinerAddressConfig
+	DAGStore   DAGStoreConfig
+}
+
+type DAGStoreConfig struct {
+	// Path to the dagstore root directory. This directory contains three
+	// subdirectories, which can be symlinked to alternative locations if
+	// need be:
+	//
+	//  - ./transients: caches unsealed deals that have been fetched from the
+	//    storage subsystem for serving retrievals.
+	//  - ./indices: stores shard indices.
+	//  - ./datastore: holds the KV store tracking the state of every shard
+	//    known to the DAG store.
+	//
+	// Default value: $LOTUS_MARKETS_PATH/dagStore (split deployment) or
+	// $LOTUS_MINER_PATH/dagStore (monolith deployment)
+	RootDir string
+
+	// The maximum amount of indexing jobs that can run simultaneously.
+	// Default value: 5.
+	MaxConcurrentIndex int
+
+	// The maximum amount of unsealed deals that can be fetched simultaneously
+	// from the storage subsystem.
+	// Default value: 2.
+	MaxConcurrentReadyFetches int
+
+	// The maximum number of simultaneous inflight API calls to the storage
+	// subsystem.
+	// Default value: 100.
+	MaxConcurrencyStorageCalls int
+
+	// The time between calls to periodic dagstore GC, in time.Duration string
+	// representation, e.g. 1m, 5m, 1h.
+	// Default value: 1 minute.
+	GCInterval Duration
 }
 
 type MinerSubsystemConfig struct {

node/impl/client/client.go

@@ -10,8 +10,7 @@ import (
 	"sort"
 	"time"
 
-	"github.com/filecoin-project/go-fil-markets/filestorecaradapter"
-	"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
+	"github.com/ipld/go-car"
 	carv2 "github.com/ipld/go-car/v2"
 	"github.com/ipld/go-car/v2/blockstore"
 
@@ -26,7 +25,7 @@ import (
 	files "github.com/ipfs/go-ipfs-files"
 	"github.com/ipfs/go-merkledag"
 	unixfile "github.com/ipfs/go-unixfs/file"
-	"github.com/ipld/go-car"
+
 	basicnode "github.com/ipld/go-ipld-prime/node/basic"
 	"github.com/ipld/go-ipld-prime/traversal/selector"
 	"github.com/ipld/go-ipld-prime/traversal/selector/builder"
@@ -47,6 +46,7 @@ import (
 	"github.com/filecoin-project/go-fil-markets/shared"
 	"github.com/filecoin-project/go-fil-markets/storagemarket"
 	"github.com/filecoin-project/go-fil-markets/storagemarket/network"
+	"github.com/filecoin-project/go-fil-markets/stores"
 	"github.com/filecoin-project/go-state-types/abi"
 	"github.com/filecoin-project/specs-actors/v3/actors/builtin/market"
 
@@ -54,6 +54,7 @@ import (
 
 	"github.com/filecoin-project/lotus/api"
 	"github.com/filecoin-project/lotus/build"
+	"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
 	"github.com/filecoin-project/lotus/chain/store"
 	"github.com/filecoin-project/lotus/chain/types"
 	"github.com/filecoin-project/lotus/markets/utils"
@@ -188,17 +189,17 @@ func (a *API) dealStarter(ctx context.Context, params *api.StartDealParams, isSt
 		providerInfo := utils.NewStorageProviderInfo(params.Miner, mi.Worker, mi.SectorSize, *mi.PeerId, mi.Multiaddrs)
 
 		result, err := a.SMDealClient.ProposeStorageDeal(ctx, storagemarket.ProposeStorageDealParams{
-			Addr:                   params.Wallet,
-			Info:                   &providerInfo,
-			Data:                   params.Data,
-			StartEpoch:             dealStart,
-			EndEpoch:               calcDealExpiration(params.MinBlocksDuration, md, dealStart),
-			Price:                  params.EpochPrice,
-			Collateral:             params.ProviderCollateral,
-			Rt:                     st,
-			FastRetrieval:          params.FastRetrieval,
-			VerifiedDeal:           params.VerifiedDeal,
-			FilestoreCARv2FilePath: CARV2FilePath,
+			Addr:          params.Wallet,
+			Info:          &providerInfo,
+			Data:          params.Data,
+			StartEpoch:    dealStart,
+			EndEpoch:      calcDealExpiration(params.MinBlocksDuration, md, dealStart),
+			Price:         params.EpochPrice,
+			Collateral:    params.ProviderCollateral,
+			Rt:            st,
+			FastRetrieval: params.FastRetrieval,
+			VerifiedDeal:  params.VerifiedDeal,
+			IndexedCAR:    CARV2FilePath,
 		})
 
 		if err != nil {
@@ -519,7 +520,7 @@ func (a *API) ClientImport(ctx context.Context, ref api.FileRef) (res *api.Impor
 			}
 		}()
 
-		root, err = a.importNormalFileToFilestoreCARv2(ctx, id, ref.Path, carFile)
+		root, err = a.doImport(ctx, ref.Path, carFile)
 		if err != nil {
 			return nil, xerrors.Errorf("failed to import normal file to CARv2: %w", err)
 		}
@@ -1031,21 +1032,21 @@ func (w *lenWriter) Write(p []byte) (n int, err error) {
 }
 
 func (a *API) ClientDealSize(ctx context.Context, root cid.Cid) (api.DataSize, error) {
-	fc, err := a.imgr().FilestoreCARV2FilePathFor(root)
+	path, err := a.imgr().FilestoreCARV2FilePathFor(root)
 	if err != nil {
 		return api.DataSize{}, xerrors.Errorf("failed to find CARv2 file for root: %w", err)
 	}
-	if len(fc) == 0 {
+	if len(path) == 0 {
 		return api.DataSize{}, xerrors.New("no CARv2 file for root")
 	}
 
-	rdOnly, err := filestorecaradapter.NewReadOnlyFileStore(fc)
+	fs, err := stores.ReadOnlyFilestore(path)
 	if err != nil {
-		return api.DataSize{}, xerrors.Errorf("failed to open read only filestore: %w", err)
+		return api.DataSize{}, xerrors.Errorf("failed to open filestore from carv2 in path %s: %w", path, err)
 	}
-	defer rdOnly.Close() //nolint:errcheck
+	defer fs.Close() //nolint:errcheck
 
-	dag := merkledag.NewDAGService(blockservice.New(rdOnly, offline.Exchange(rdOnly)))
+	dag := merkledag.NewDAGService(blockservice.New(fs, offline.Exchange(fs)))
 	w := lenWriter(0)
 
 	err = car.WriteCar(ctx, dag, []cid.Cid{root}, &w)
@@ -1062,21 +1063,21 @@ func (a *API) ClientDealSize(ctx context.Context, root cid.Cid) (api.DataSize, e
 }
 
 func (a *API) ClientDealPieceCID(ctx context.Context, root cid.Cid) (api.DataCIDSize, error) {
-	fc, err := a.imgr().FilestoreCARV2FilePathFor(root)
+	path, err := a.imgr().FilestoreCARV2FilePathFor(root)
 	if err != nil {
 		return api.DataCIDSize{}, xerrors.Errorf("failed to find CARv2 file for root: %w", err)
 	}
-	if len(fc) == 0 {
+	if len(path) == 0 {
 		return api.DataCIDSize{}, xerrors.New("no CARv2 file for root")
 	}
 
-	rdOnly, err := filestorecaradapter.NewReadOnlyFileStore(fc)
+	fs, err := stores.ReadOnlyFilestore(path)
 	if err != nil {
-		return api.DataCIDSize{}, xerrors.Errorf("failed to open read only blockstore: %w", err)
+		return api.DataCIDSize{}, xerrors.Errorf("failed to open filestore from carv2 in path %s: %w", path, err)
 	}
-	defer rdOnly.Close() //nolint:errcheck
+	defer fs.Close() //nolint:errcheck
 
-	dag := merkledag.NewDAGService(blockservice.New(rdOnly, offline.Exchange(rdOnly)))
+	dag := merkledag.NewDAGService(blockservice.New(fs, offline.Exchange(fs)))
 	w := &writer.Writer{}
 	bw := bufio.NewWriterSize(w, int(writer.CommPBuf))
 
@@ -1095,22 +1096,20 @@ func (a *API) ClientDealPieceCID(ctx context.Context, root cid.Cid) (api.DataCID
 
 func (a *API) ClientGenCar(ctx context.Context, ref api.FileRef, outputPath string) error {
 	id := importmgr.ImportID(rand.Uint64())
-	tmpCARv2File, err := a.imgr().NewTempFile(id)
+	tmp, err := a.imgr().NewTempFile(id)
 	if err != nil {
 		return xerrors.Errorf("failed to create temp file: %w", err)
 	}
-	defer os.Remove(tmpCARv2File) //nolint:errcheck
+	defer os.Remove(tmp) //nolint:errcheck
 
-	root, err := a.importNormalFileToFilestoreCARv2(ctx, id, ref.Path, tmpCARv2File)
+	root, err := a.doImport(ctx, ref.Path, tmp)
 	if err != nil {
 		return xerrors.Errorf("failed to import normal file to CARv2")
 	}
 
-	// generate a deterministic CARv1 payload from the UnixFS DAG by doing an IPLD
-	// traversal over the Unixfs DAG in the CARv2 file using the "all selector" i.e the entire DAG selector.
-	fs, err := filestorecaradapter.NewReadOnlyFileStore(tmpCARv2File)
+	fs, err := stores.ReadOnlyFilestore(tmp)
 	if err != nil {
-		return xerrors.Errorf("failed to open read only CARv2 blockstore: %w", err)
+		return xerrors.Errorf("failed to open filestore from carv2 in path %s: %w", tmp, err)
 	}
 	defer fs.Close() //nolint:errcheck
 

node/impl/client/import.go

@@ -2,18 +2,13 @@ package client
 
 import (
 	"context"
+	"io/ioutil"
 	"os"
 
-	"github.com/filecoin-project/go-fil-markets/filestorecaradapter"
-	bstore "github.com/filecoin-project/lotus/blockstore"
-	"github.com/filecoin-project/lotus/build"
-	"github.com/filecoin-project/lotus/node/repo/importmgr"
+	"github.com/filecoin-project/go-fil-markets/stores"
 	"github.com/ipfs/go-blockservice"
 	"github.com/ipfs/go-cid"
 	"github.com/ipfs/go-cidutil"
-	"github.com/ipfs/go-datastore"
-	ds_sync "github.com/ipfs/go-datastore/sync"
-	"github.com/ipfs/go-filestore"
 	chunker "github.com/ipfs/go-ipfs-chunker"
 	offline "github.com/ipfs/go-ipfs-exchange-offline"
 	files2 "github.com/ipfs/go-ipfs-files"
@@ -21,56 +16,83 @@ import (
 	"github.com/ipfs/go-merkledag"
 	"github.com/ipfs/go-unixfs/importer/balanced"
 	ihelper "github.com/ipfs/go-unixfs/importer/helpers"
+	"github.com/ipld/go-car/v2"
+	"github.com/ipld/go-car/v2/blockstore"
 	"golang.org/x/xerrors"
+
+	"github.com/filecoin-project/lotus/build"
 )
 
-// importNormalFileToFilestoreCARv2 transforms the client's "normal file" to a Unixfs IPLD DAG and writes out the DAG to a CARv2 file
-// that can be used to back a filestore.
-func (a *API) importNormalFileToFilestoreCARv2(ctx context.Context, importID importmgr.ImportID, inputFilePath string, outputCARv2Path string) (c cid.Cid, finalErr error) {
-
-	// TODO: We've currently put in a hack to create the Unixfs DAG as a CARv2 without using Badger.
-	// We first create the Unixfs DAG using a filestore to get the root of the Unixfs DAG.
-	// We can't create the UnixfsDAG right away using a CARv2 read-write blockstore as the blockstore
-	// needs the root of the DAG during instantiation to write out a valid CARv2 file.
+// doImport takes a standard file (src), forms a UnixFS DAG, and writes a
+// CARv2 file with positional mapping (backed by the go-filestore library).
+func (a *API) doImport(ctx context.Context, src string, dst string) (cid.Cid, error) {
+	// This method uses a two-phase approach with a staging CAR blockstore and
+	// a final CAR blockstore.
 	//
-	// In the second pass, we create a CARv2 file with the root present using the root node we get in the above step.
-	// This hack should be fixed when CARv2 allows specifying the root AFTER finishing the CARv2 streaming write.
-	fm := filestore.NewFileManager(ds_sync.MutexWrap(datastore.NewMapDatastore()), "/")
-	fm.AllowFiles = true
-	fstore := filestore.NewFilestore(bstore.NewMemorySync(), fm)
+	// This is necessary because of https://github.com/ipld/go-car/issues/196
+	//
+	// TODO: do we need to chunk twice? Isn't the first output already in the
+	//  right order? Can't we just copy the CAR file and replace the header?
+	f, err := ioutil.TempFile("", "")
+	if err != nil {
+		return cid.Undef, xerrors.Errorf("failed to create temp file: %w", err)
+	}
+	_ = f.Close() // close; we only want the path.
+	tmp := f.Name()
+	defer os.Remove(tmp) //nolint:errcheck
+
+	// Step 1. Compute the UnixFS DAG and write it to a CARv2 file to get
+	// the root CID of the DAG.
+	fstore, err := stores.ReadWriteFilestore(tmp)
+	if err != nil {
+		return cid.Undef, xerrors.Errorf("failed to create temporary filestore: %w", err)
+	}
+
 	bsvc := blockservice.New(fstore, offline.Exchange(fstore))
-	defer bsvc.Close() //nolint:errcheck
+	dags := merkledag.NewDAGService(bsvc)
 
-	// ---- First Pass ---  Write out the UnixFS DAG to a rootless CARv2 file by instantiating a read-write CARv2 blockstore without the root.
-	root, err := importNormalFileToUnixfsDAG(ctx, inputFilePath, merkledag.NewDAGService(bsvc))
+	root, err := buildUnixFS(ctx, src, dags)
 	if err != nil {
-		return cid.Undef, xerrors.Errorf("failed to import file to store: %w", err)
+		_ = fstore.Close()
+		return cid.Undef, xerrors.Errorf("failed to import file to store to compute root: %w", err)
 	}
 
-	//------ Second Pass --- Now that we have the root of the Unixfs DAG -> write out the Unixfs DAG to a CARv2 file with the root present by using a
-	// filestore backed by a read-write CARv2 blockstore.
-	fsb, err := filestorecaradapter.NewReadWriteFileStore(outputCARv2Path, []cid.Cid{root})
-	if err != nil {
-		return cid.Undef, xerrors.Errorf("failed to create a CARv2 read-write blockstore: %w", err)
-	}
-	defer fsb.Close() //nolint:errcheck
-
-	bsvc = blockservice.New(fsb, offline.Exchange(fsb))
-	root2, err := importNormalFileToUnixfsDAG(ctx, inputFilePath, merkledag.NewDAGService(bsvc))
-	if err != nil {
-		return cid.Undef, xerrors.Errorf("failed to create Unixfs DAG with CARv2 blockstore: %w", err)
+	if err := fstore.Close(); err != nil {
+		return cid.Undef, xerrors.Errorf("failed to finalize car filestore: %w", err)
 	}
 
-	if root != root2 {
+	// Step 2. We now have the root of the UnixFS DAG, and we can write the
+	// final CAR for real under `dst`.
+	bs, err := blockstore.OpenReadWrite(dst, []cid.Cid{root},
+		car.ZeroLengthSectionAsEOF(true),
+		blockstore.UseWholeCIDs(true))
+	if err != nil {
+		return cid.Undef, xerrors.Errorf("failed to create a carv2 read/write blockstore: %w", err)
+	}
+
+	bsvc = blockservice.New(bs, offline.Exchange(bs))
+	dags = merkledag.NewDAGService(bsvc)
+	finalRoot, err := buildUnixFS(ctx, src, dags)
+	if err != nil {
+		_ = bs.Close()
+		return cid.Undef, xerrors.Errorf("failed to create UnixFS DAG with carv2 blockstore: %w", err)
+	}
+
+	if err := bs.Finalize(); err != nil {
+		return cid.Undef, xerrors.Errorf("failed to finalize car blockstore: %w", err)
+	}
+
+	if root != finalRoot {
 		return cid.Undef, xerrors.New("roots do not match")
 	}
 
 	return root, nil
 }
 
-// importNormalFileToUnixfsDAG transforms a client's normal file to a UnixfsDAG and imports the DAG to the given DAG service.
-func importNormalFileToUnixfsDAG(ctx context.Context, inputFilePath string, dag ipld.DAGService) (cid.Cid, error) {
-	f, err := os.Open(inputFilePath)
+// buildUnixFS builds a UnixFS DAG out of the supplied ordinary file,
+// and imports the DAG into the supplied service.
+func buildUnixFS(ctx context.Context, src string, dag ipld.DAGService) (cid.Cid, error) {
+	f, err := os.Open(src)
 	if err != nil {
 		return cid.Undef, xerrors.Errorf("failed to open input file: %w", err)
 	}
@@ -81,7 +103,7 @@ func importNormalFileToUnixfsDAG(ctx context.Context, inputFilePath string, dag
 		return cid.Undef, xerrors.Errorf("failed to stat file :%w", err)
 	}
 
-	file, err := files2.NewReaderPathFile(inputFilePath, f, stat)
+	file, err := files2.NewReaderPathFile(src, f, stat)
 	if err != nil {
 		return cid.Undef, xerrors.Errorf("failed to create reader path file: %w", err)
 	}

node/impl/client/import_test.go

@@ -4,12 +4,11 @@ import (
 	"context"
 	"io"
 	"io/ioutil"
-	"math/rand"
 	"os"
 	"strings"
 	"testing"
 
-	"github.com/filecoin-project/go-fil-markets/filestorecaradapter"
+	"github.com/filecoin-project/go-fil-markets/stores"
 	"github.com/filecoin-project/lotus/node/repo/importmgr"
 	"github.com/ipfs/go-blockservice"
 	"github.com/ipfs/go-cid"
@@ -22,39 +21,50 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func TestImportNormalFileToUnixfsDAG(t *testing.T) {
+// This test uses a full "dense" CARv2, and not a filestore (positional mapping).
+func TestRoundtripUnixFS_Dense(t *testing.T) {
 	ctx := context.Background()
-	inputPath, inputContents := genNormalInputFile(t)
+
+	inputPath, inputContents := genInputFile(t)
 	defer os.Remove(inputPath) //nolint:errcheck
-	carv2File := genTmpFile(t)
+
+	carv2File := newTmpFile(t)
 	defer os.Remove(carv2File) //nolint:errcheck
 
-	// import a normal file to a Unixfs DAG using a CARv2 read-write blockstore and flush it out to a CARv2 file.
-	tempCARv2Store, err := blockstore.OpenReadWrite(carv2File, []cid.Cid{}, carv2.ZeroLengthSectionAsEOF(true), blockstore.UseWholeCIDs(true))
+	// import a file to a Unixfs DAG using a CARv2 read/write blockstore.
+	path, err := blockstore.OpenReadWrite(carv2File, nil,
+		carv2.ZeroLengthSectionAsEOF(true),
+		blockstore.UseWholeCIDs(true))
 	require.NoError(t, err)
-	bsvc := blockservice.New(tempCARv2Store, offline.Exchange(tempCARv2Store))
-	root, err := importNormalFileToUnixfsDAG(ctx, inputPath, merkledag.NewDAGService(bsvc))
+
+	bsvc := blockservice.New(path, offline.Exchange(path))
+	dags := merkledag.NewDAGService(bsvc)
+
+	root, err := buildUnixFS(ctx, inputPath, dags)
 	require.NoError(t, err)
 	require.NotEqual(t, cid.Undef, root)
-	require.NoError(t, tempCARv2Store.Finalize())
+	require.NoError(t, path.Finalize())
 
-	// convert the CARv2 file to a normal file again and ensure the contents match.
-	readOnly, err := blockstore.OpenReadOnly(carv2File, carv2.ZeroLengthSectionAsEOF(true), blockstore.UseWholeCIDs(true))
+	// reconstruct the file.
+	readOnly, err := blockstore.OpenReadOnly(carv2File,
+		carv2.ZeroLengthSectionAsEOF(true),
+		blockstore.UseWholeCIDs(true))
 	require.NoError(t, err)
 	defer readOnly.Close() //nolint:errcheck
-	dag := merkledag.NewDAGService(blockservice.New(readOnly, offline.Exchange(readOnly)))
 
-	nd, err := dag.Get(ctx, root)
-	require.NoError(t, err)
-	file, err := unixfile.NewUnixfsFile(ctx, dag, nd)
+	dags = merkledag.NewDAGService(blockservice.New(readOnly, offline.Exchange(readOnly)))
+
+	nd, err := dags.Get(ctx, root)
 	require.NoError(t, err)
 
-	tmpOutput := genTmpFile(t)
+	file, err := unixfile.NewUnixfsFile(ctx, dags, nd)
+	require.NoError(t, err)
+
+	tmpOutput := newTmpFile(t)
 	defer os.Remove(tmpOutput) //nolint:errcheck
 	require.NoError(t, files.WriteTo(file, tmpOutput))
 
 	// ensure contents of the initial input file and the output file are identical.
-
 	fo, err := os.Open(tmpOutput)
 	require.NoError(t, err)
 	bz2, err := ioutil.ReadAll(fo)
@@ -63,35 +73,36 @@ func TestImportNormalFileToUnixfsDAG(t *testing.T) {
 	require.Equal(t, inputContents, bz2)
 }
 
-func TestImportNormalFileToCARv2(t *testing.T) {
+func TestRoundtripUnixFS_Filestore(t *testing.T) {
 	ctx := context.Background()
 	a := &API{
 		Imports: &importmgr.Mgr{},
 	}
-	importID := importmgr.ImportID(rand.Uint64())
 
-	inputFilePath, inputContents := genNormalInputFile(t)
+	inputFilePath, inputContents := genInputFile(t)
 	defer os.Remove(inputFilePath) //nolint:errcheck
 
-	outputCARv2 := genTmpFile(t)
-	defer os.Remove(outputCARv2) //nolint:errcheck
+	path := newTmpFile(t)
+	defer os.Remove(path) //nolint:errcheck
 
-	root, err := a.importNormalFileToFilestoreCARv2(ctx, importID, inputFilePath, outputCARv2)
+	root, err := a.doImport(ctx, inputFilePath, path)
 	require.NoError(t, err)
 	require.NotEqual(t, cid.Undef, root)
 
 	// convert the CARv2 to a normal file again and ensure the contents match
-	readOnly, err := filestorecaradapter.NewReadOnlyFileStore(outputCARv2)
+	fs, err := stores.ReadOnlyFilestore(path)
 	require.NoError(t, err)
-	defer readOnly.Close() //nolint:errcheck
-	dag := merkledag.NewDAGService(blockservice.New(readOnly, offline.Exchange(readOnly)))
+	defer fs.Close() //nolint:errcheck
 
-	nd, err := dag.Get(ctx, root)
-	require.NoError(t, err)
-	file, err := unixfile.NewUnixfsFile(ctx, dag, nd)
+	dags := merkledag.NewDAGService(blockservice.New(fs, offline.Exchange(fs)))
+
+	nd, err := dags.Get(ctx, root)
 	require.NoError(t, err)
 
-	tmpOutput := genTmpFile(t)
+	file, err := unixfile.NewUnixfsFile(ctx, dags, nd)
+	require.NoError(t, err)
+
+	tmpOutput := newTmpFile(t)
 	defer os.Remove(tmpOutput) //nolint:errcheck
 	require.NoError(t, files.WriteTo(file, tmpOutput))
 
@@ -104,14 +115,14 @@ func TestImportNormalFileToCARv2(t *testing.T) {
 	require.Equal(t, inputContents, bz2)
 }
 
-func genTmpFile(t *testing.T) string {
+func newTmpFile(t *testing.T) string {
 	f, err := os.CreateTemp("", "")
 	require.NoError(t, err)
 	require.NoError(t, f.Close())
 	return f.Name()
 }
 
-func genNormalInputFile(t *testing.T) (filepath string, contents []byte) {
+func genInputFile(t *testing.T) (filepath string, contents []byte) {
 	s := strings.Repeat("abcde", 100)
 	tmp, err := os.CreateTemp("", "")
 	require.NoError(t, err)

node/impl/storminer.go

@@ -3,16 +3,20 @@ package impl
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"net/http"
 	"os"
+	"sort"
 	"strconv"
 	"time"
 
+	"github.com/filecoin-project/dagstore"
+	"github.com/filecoin-project/dagstore/shard"
 	"github.com/filecoin-project/go-jsonrpc/auth"
+
 	"github.com/filecoin-project/lotus/chain/actors/builtin"
 	"github.com/filecoin-project/lotus/chain/gen"
 
-	"github.com/filecoin-project/lotus/build"
 	"github.com/google/uuid"
 	"github.com/ipfs/go-cid"
 	"github.com/libp2p/go-libp2p-core/host"
@@ -20,6 +24,8 @@ import (
 	"go.uber.org/fx"
 	"golang.org/x/xerrors"
 
+	"github.com/filecoin-project/lotus/build"
+
 	"github.com/filecoin-project/go-address"
 	datatransfer "github.com/filecoin-project/go-data-transfer"
 	"github.com/filecoin-project/go-fil-markets/piecestore"
@@ -34,6 +40,8 @@ import (
 	sealing "github.com/filecoin-project/lotus/extern/storage-sealing"
 	"github.com/filecoin-project/lotus/extern/storage-sealing/sealiface"
 
+	sto "github.com/filecoin-project/specs-storage/storage"
+
 	"github.com/filecoin-project/lotus/api"
 	apitypes "github.com/filecoin-project/lotus/api/types"
 	"github.com/filecoin-project/lotus/chain/types"
@@ -42,7 +50,6 @@ import (
 	"github.com/filecoin-project/lotus/node/modules/dtypes"
 	"github.com/filecoin-project/lotus/storage"
 	"github.com/filecoin-project/lotus/storage/sectorblocks"
-	sto "github.com/filecoin-project/specs-storage/storage"
 )
 
 type StorageMinerAPI struct {
@@ -65,6 +72,7 @@ type StorageMinerAPI struct {
 	DealPublisher     *storageadapter.DealPublisher     `optional:"true"`
 	SectorBlocks      *sectorblocks.SectorBlocks        `optional:"true"`
 	Host              host.Host                         `optional:"true"`
+	DAGStore          *dagstore.DAGStore                `optional:"true"`
 
 	// Miner / storage
 	Miner       *storage.Miner              `optional:"true"`
@@ -98,6 +106,8 @@ type StorageMinerAPI struct {
 	SetExpectedSealDurationFunc                 dtypes.SetExpectedSealDurationFunc                 `optional:"true"`
 }
 
+var _ api.StorageMiner = &StorageMinerAPI{}
+
 func (sm *StorageMinerAPI) ServeRemote(perm bool) func(w http.ResponseWriter, r *http.Request) {
 	return func(w http.ResponseWriter, r *http.Request) {
 		if perm == true {
@@ -545,6 +555,225 @@ func (sm *StorageMinerAPI) MarketPublishPendingDeals(ctx context.Context) error
 	return nil
 }
 
+func (sm *StorageMinerAPI) DagstoreListShards(ctx context.Context) ([]api.DagstoreShardInfo, error) {
+	if sm.DAGStore == nil {
+		return nil, fmt.Errorf("dagstore not available on this node")
+	}
+
+	info := sm.DAGStore.AllShardsInfo()
+	ret := make([]api.DagstoreShardInfo, 0, len(info))
+	for k, i := range info {
+		ret = append(ret, api.DagstoreShardInfo{
+			Key:   k.String(),
+			State: i.ShardState.String(),
+			Error: func() string {
+				if i.Error == nil {
+					return ""
+				}
+				return i.Error.Error()
+			}(),
+		})
+	}
+
+	// order by key.
+	sort.SliceStable(ret, func(i, j int) bool {
+		return ret[i].Key < ret[j].Key
+	})
+
+	return ret, nil
+}
+
+func (sm *StorageMinerAPI) DagstoreInitializeShard(ctx context.Context, key string) error {
+	if sm.DAGStore == nil {
+		return fmt.Errorf("dagstore not available on this node")
+	}
+
+	k := shard.KeyFromString(key)
+
+	info, err := sm.DAGStore.GetShardInfo(k)
+	if err != nil {
+		return fmt.Errorf("failed to get shard info: %w", err)
+	}
+	if st := info.ShardState; st != dagstore.ShardStateNew {
+		return fmt.Errorf("cannot initialize shard; expected state ShardStateNew, was: %s", st.String())
+	}
+
+	ch := make(chan dagstore.ShardResult, 1)
+	if err = sm.DAGStore.AcquireShard(ctx, k, ch, dagstore.AcquireOpts{}); err != nil {
+		return fmt.Errorf("failed to acquire shard: %w", err)
+	}
+
+	var res dagstore.ShardResult
+	select {
+	case res = <-ch:
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+
+	if err := res.Error; err != nil {
+		return fmt.Errorf("failed to acquire shard: %w", err)
+	}
+
+	if res.Accessor != nil {
+		err = res.Accessor.Close()
+		if err != nil {
+			log.Warnw("failed to close shard accessor; continuing", "shard_key", k, "error", err)
+		}
+	}
+
+	return nil
+}
+
+func (sm *StorageMinerAPI) DagstoreInitializeAll(ctx context.Context, params api.DagstoreInitializeAllParams) (<-chan api.DagstoreInitializeAllEvent, error) {
+	if sm.DAGStore == nil {
+		return nil, fmt.Errorf("dagstore not available on this node")
+	}
+
+	// prepare the thottler tokens.
+	var throttle chan struct{}
+	if c := params.MaxConcurrency; c > 0 {
+		throttle = make(chan struct{}, c)
+		for i := 0; i < c; i++ {
+			throttle <- struct{}{}
+		}
+	}
+
+	info := sm.DAGStore.AllShardsInfo()
+	var uninit []string
+	for k, i := range info {
+		if i.ShardState != dagstore.ShardStateNew {
+			continue
+		}
+		uninit = append(uninit, k.String())
+	}
+
+	total := len(uninit)
+	if total == 0 {
+		out := make(chan api.DagstoreInitializeAllEvent)
+		close(out)
+		return out, nil
+	}
+
+	// response channel must be closed when we're done, or the context is cancelled.
+	// this buffering is necessary to prevent inflight children goroutines from
+	// publishing to a closed channel (res) when the context is cancelled.
+	out := make(chan api.DagstoreInitializeAllEvent, 32) // internal buffer.
+	res := make(chan api.DagstoreInitializeAllEvent, 32) // returned to caller.
+
+	// pump events back to caller.
+	// two events per shard.
+	go func() {
+		defer close(res)
+
+		for i := 0; i < total*2; i++ {
+			select {
+			case res <- <-out:
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
+	go func() {
+		for i, k := range uninit {
+			select {
+			case <-throttle:
+				// acquired a throttle token, proceed.
+			case <-ctx.Done():
+				return
+			}
+
+			go func(k string, i int) {
+				r := api.DagstoreInitializeAllEvent{
+					Key:     k,
+					Event:   "start",
+					Total:   total,
+					Current: i + 1, // start with 1
+				}
+				select {
+				case out <- r:
+				case <-ctx.Done():
+					return
+				}
+
+				err := sm.DagstoreInitializeShard(ctx, k)
+				throttle <- struct{}{}
+
+				r.Event = "end"
+				if err == nil {
+					r.Success = true
+				} else {
+					r.Success = false
+					r.Error = err.Error()
+				}
+
+				select {
+				case out <- r:
+				case <-ctx.Done():
+				}
+			}(k, i)
+		}
+	}()
+
+	return res, nil
+
+}
+
+func (sm *StorageMinerAPI) DagstoreRecoverShard(ctx context.Context, key string) error {
+	if sm.DAGStore == nil {
+		return fmt.Errorf("dagstore not available on this node")
+	}
+
+	k := shard.KeyFromString(key)
+
+	info, err := sm.DAGStore.GetShardInfo(k)
+	if err != nil {
+		return fmt.Errorf("failed to get shard info: %w", err)
+	}
+	if st := info.ShardState; st != dagstore.ShardStateErrored {
+		return fmt.Errorf("cannot recover shard; expected state ShardStateErrored, was: %s", st.String())
+	}
+
+	ch := make(chan dagstore.ShardResult, 1)
+	if err = sm.DAGStore.RecoverShard(ctx, k, ch, dagstore.RecoverOpts{}); err != nil {
+		return fmt.Errorf("failed to recover shard: %w", err)
+	}
+
+	var res dagstore.ShardResult
+	select {
+	case res = <-ch:
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+
+	return res.Error
+}
+
+func (sm *StorageMinerAPI) DagstoreGC(ctx context.Context) ([]api.DagstoreShardResult, error) {
+	if sm.DAGStore == nil {
+		return nil, fmt.Errorf("dagstore not available on this node")
+	}
+
+	res, err := sm.DAGStore.GC(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to gc: %w", err)
+	}
+
+	ret := make([]api.DagstoreShardResult, 0, len(res.Shards))
+	for k, err := range res.Shards {
+		r := api.DagstoreShardResult{Key: k.String()}
+		if err == nil {
+			r.Success = true
+		} else {
+			r.Success = false
+			r.Error = err.Error()
+		}
+		ret = append(ret, r)
+	}
+
+	return ret, nil
+}
+
 func (sm *StorageMinerAPI) DealsList(ctx context.Context) ([]api.MarketDeal, error) {
 	return sm.listDeals(ctx)
 }
@@ -708,5 +937,3 @@ func (sm *StorageMinerAPI) ComputeProof(ctx context.Context, ssi []builtin.Secto
 func (sm *StorageMinerAPI) RuntimeSubsystems(context.Context) (res api.MinerSubsystems, err error) {
 	return sm.EnabledSubsystems, nil
 }
-
-var _ api.StorageMiner = &StorageMinerAPI{}

node/modules/client.go

@@ -182,7 +182,7 @@ func StorageClient(lc fx.Lifecycle, h host.Host, dataTransfer dtypes.ClientDataT
 func RetrievalClient(lc fx.Lifecycle, h host.Host, r repo.LockedRepo, dt dtypes.ClientDataTransfer, payAPI payapi.PaychAPI, resolver discovery.PeerResolver,
 	ds dtypes.MetadataDS, chainAPI full.ChainAPI, stateAPI full.StateAPI, j journal.Journal) (retrievalmarket.RetrievalClient, error) {
 
-	carsPath := filepath.Join(r.Path(), dagStore, "retrieval-cars")
+	carsPath := filepath.Join(r.Path(), DefaultDAGStoreDir, "retrieval-cars")
 
 	if err := os.MkdirAll(carsPath, 0755); err != nil {
 		return nil, xerrors.Errorf("failed to create dir")

node/modules/storageminer.go

@@ -7,7 +7,6 @@ import (
 	"fmt"
 	"net/http"
 	"os"
-	"path"
 	"path/filepath"
 	"strings"
 	"time"
@@ -16,26 +15,28 @@ import (
 	"go.uber.org/multierr"
 	"golang.org/x/xerrors"
 
+	"github.com/filecoin-project/go-address"
+	dtimpl "github.com/filecoin-project/go-data-transfer/impl"
+	dtnet "github.com/filecoin-project/go-data-transfer/network"
+	dtgstransport "github.com/filecoin-project/go-data-transfer/transport/graphsync"
+	"github.com/filecoin-project/go-jsonrpc/auth"
+	"github.com/filecoin-project/go-paramfetch"
+	"github.com/filecoin-project/go-state-types/abi"
+	"github.com/filecoin-project/go-statestore"
+	"github.com/filecoin-project/go-storedcounter"
 	"github.com/ipfs/go-bitswap"
 	"github.com/ipfs/go-bitswap/network"
 	"github.com/ipfs/go-blockservice"
 	"github.com/ipfs/go-cid"
 	"github.com/ipfs/go-datastore"
 	"github.com/ipfs/go-datastore/namespace"
-	levelds "github.com/ipfs/go-ds-leveldb"
-	measure "github.com/ipfs/go-ds-measure"
 	graphsync "github.com/ipfs/go-graphsync/impl"
 	gsnet "github.com/ipfs/go-graphsync/network"
 	"github.com/ipfs/go-graphsync/storeutil"
 	"github.com/ipfs/go-merkledag"
 	"github.com/libp2p/go-libp2p-core/host"
 	"github.com/libp2p/go-libp2p-core/routing"
-	ldbopts "github.com/syndtr/goleveldb/leveldb/opt"
 
-	"github.com/filecoin-project/go-address"
-	dtimpl "github.com/filecoin-project/go-data-transfer/impl"
-	dtnet "github.com/filecoin-project/go-data-transfer/network"
-	dtgstransport "github.com/filecoin-project/go-data-transfer/transport/graphsync"
 	piecefilestore "github.com/filecoin-project/go-fil-markets/filestore"
 	piecestoreimpl "github.com/filecoin-project/go-fil-markets/piecestore/impl"
 	"github.com/filecoin-project/go-fil-markets/retrievalmarket"
@@ -46,11 +47,6 @@ import (
 	storageimpl "github.com/filecoin-project/go-fil-markets/storagemarket/impl"
 	"github.com/filecoin-project/go-fil-markets/storagemarket/impl/storedask"
 	smnet "github.com/filecoin-project/go-fil-markets/storagemarket/network"
-	"github.com/filecoin-project/go-jsonrpc/auth"
-	"github.com/filecoin-project/go-paramfetch"
-	"github.com/filecoin-project/go-state-types/abi"
-	"github.com/filecoin-project/go-statestore"
-	"github.com/filecoin-project/go-storedcounter"
 
 	sectorstorage "github.com/filecoin-project/lotus/extern/sector-storage"
 	"github.com/filecoin-project/lotus/extern/sector-storage/ffiwrapper"
@@ -81,7 +77,6 @@ import (
 )
 
 var StorageCounterDSPrefix = "/storage/nextid"
-var dagStore = "dagStore"
 
 func minerAddrFromDS(ds dtypes.MetadataDS) (address.Address, error) {
 	maddrb, err := ds.Get(datastore.NewKey("miner-address"))
@@ -580,91 +575,6 @@ func BasicDealFilter(user dtypes.StorageDealFilter) func(onlineOk dtypes.Conside
 	}
 }
 
-func NewLotusAccessor(lc fx.Lifecycle,
-	pieceStore dtypes.ProviderPieceStore,
-	rpn retrievalmarket.RetrievalProviderNode,
-) (dagstore.LotusAccessor, error) {
-	mountApi := dagstore.NewLotusAccessor(pieceStore, rpn)
-	ready := make(chan error, 1)
-	pieceStore.OnReady(func(err error) {
-		ready <- err
-	})
-	lc.Append(fx.Hook{
-		OnStart: func(ctx context.Context) error {
-			if err := <-ready; err != nil {
-				return fmt.Errorf("aborting dagstore start; piecestore failed to start: %s", err)
-			}
-			return mountApi.Start(ctx)
-		},
-		OnStop: func(context.Context) error {
-			return nil
-		},
-	})
-
-	return mountApi, nil
-}
-
-func DagStoreWrapper(
-	lc fx.Lifecycle,
-	r repo.LockedRepo,
-	lotusAccessor dagstore.LotusAccessor,
-) (*dagstore.Wrapper, error) {
-	dagStoreDir := filepath.Join(r.Path(), dagStore)
-	dagStoreDS, err := createDAGStoreDatastore(dagStoreDir)
-	if err != nil {
-		return nil, err
-	}
-
-	cfg := dagstore.MarketDAGStoreConfig{
-		TransientsDir:       filepath.Join(dagStoreDir, "transients"),
-		IndexDir:            filepath.Join(dagStoreDir, "index"),
-		Datastore:           dagStoreDS,
-		GCInterval:          1 * time.Minute,
-		MaxConcurrentIndex:  5,
-		MaxConcurrentCopies: 2,
-	}
-
-	dsw, err := dagstore.NewDagStoreWrapper(cfg, lotusAccessor)
-	if err != nil {
-		return nil, xerrors.Errorf("failed to create DAG store wrapper: %w", err)
-	}
-
-	lc.Append(fx.Hook{
-		OnStart: func(ctx context.Context) error {
-			return dsw.Start(ctx)
-		},
-		OnStop: func(context.Context) error {
-			return dsw.Close()
-		},
-	})
-	return dsw, nil
-}
-
-// createDAGStoreDatastore creates a datastore under the given base directory
-// for DAG store metadata
-func createDAGStoreDatastore(baseDir string) (datastore.Batching, error) {
-	// Create a datastore directory under the base dir if it doesn't already exist
-	dsDir := path.Join(baseDir, "datastore")
-	if err := os.MkdirAll(dsDir, 0755); err != nil {
-		return nil, xerrors.Errorf("failed to create directory %s for DAG store datastore: %w", dsDir, err)
-	}
-
-	// Create a new LevelDB datastore
-	ds, err := levelds.NewDatastore(dsDir, &levelds.Options{
-		Compression: ldbopts.NoCompression,
-		NoSync:      false,
-		Strict:      ldbopts.StrictAll,
-		ReadOnly:    false,
-	})
-	if err != nil {
-		return nil, xerrors.Errorf("failed to open datastore for DAG store: %w", err)
-	}
-
-	// Keep statistics about the datastore
-	mds := measure.New("measure.", ds)
-	return mds, nil
-}
-
 func StorageProvider(minerAddress dtypes.MinerAddress,
 	storedAsk *storedask.StoredAsk,
 	h host.Host, ds dtypes.MetadataDS,
@@ -676,18 +586,33 @@ func StorageProvider(minerAddress dtypes.MinerAddress,
 	dsw *dagstore.Wrapper,
 ) (storagemarket.StorageProvider, error) {
 	net := smnet.NewFromLibp2pHost(h)
-	store, err := piecefilestore.NewLocalFileStore(piecefilestore.OsPath(r.Path()))
+
+	// Create a directory under the repo path for temporary files
+	repoTmpPath := filepath.Join(r.Path(), "tmp")
+	err := os.MkdirAll(repoTmpPath, 0755) //nolint: gosec
+	if err != nil {
+		return nil, xerrors.Errorf("creating tmp dir for storage provider repo: %w", err)
+	}
+
+	store, err := piecefilestore.NewLocalFileStore(piecefilestore.OsPath(repoTmpPath))
 	if err != nil {
 		return nil, err
 	}
 
-	dagStorePath := filepath.Join(r.Path(), dagStore)
-
 	opt := storageimpl.CustomDealDecisionLogic(storageimpl.DealDeciderFunc(df))
-	shardMigrator := storageimpl.NewShardMigrator(address.Address(minerAddress), dagStorePath, dsw, pieceStore, spn)
 
-	return storageimpl.NewProvider(net, namespace.Wrap(ds, datastore.NewKey("/deals/provider")), store, dsw, pieceStore,
-		dataTransfer, spn, address.Address(minerAddress), storedAsk, shardMigrator, opt)
+	return storageimpl.NewProvider(
+		net,
+		namespace.Wrap(ds, datastore.NewKey("/deals/provider")),
+		store,
+		dsw,
+		pieceStore,
+		dataTransfer,
+		spn,
+		address.Address(minerAddress),
+		storedAsk,
+		opt,
+	)
 }
 
 func RetrievalDealFilter(userFilter dtypes.RetrievalDealFilter) func(onlineOk dtypes.ConsiderOnlineRetrievalDealsConfigFunc,

node/modules/storageminer_dagstore.go

@@ -0,0 +1,112 @@
+package modules
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strconv"
+
+	"github.com/filecoin-project/dagstore"
+	"go.uber.org/fx"
+	"golang.org/x/xerrors"
+
+	"github.com/filecoin-project/go-fil-markets/retrievalmarket"
+
+	mdagstore "github.com/filecoin-project/lotus/markets/dagstore"
+	"github.com/filecoin-project/lotus/node/config"
+	"github.com/filecoin-project/lotus/node/modules/dtypes"
+	"github.com/filecoin-project/lotus/node/repo"
+)
+
+const (
+	EnvDAGStoreCopyConcurrency = "LOTUS_DAGSTORE_COPY_CONCURRENCY"
+	DefaultDAGStoreDir         = "dagstore"
+)
+
+// NewMinerAPI creates a new MinerAPI adaptor for the dagstore mounts.
+func NewMinerAPI(lc fx.Lifecycle, r repo.LockedRepo, pieceStore dtypes.ProviderPieceStore, rpn retrievalmarket.RetrievalProviderNode) (mdagstore.MinerAPI, error) {
+	cfg, err := extractDAGStoreConfig(r)
+	if err != nil {
+		return nil, err
+	}
+
+	// caps the amount of concurrent calls to the storage, so that we don't
+	// spam it during heavy processes like bulk migration.
+	if v, ok := os.LookupEnv("LOTUS_DAGSTORE_MOUNT_CONCURRENCY"); ok {
+		concurrency, err := strconv.Atoi(v)
+		if err == nil {
+			cfg.MaxConcurrencyStorageCalls = concurrency
+		}
+	}
+
+	mountApi := mdagstore.NewMinerAPI(pieceStore, rpn, cfg.MaxConcurrencyStorageCalls)
+	ready := make(chan error, 1)
+	pieceStore.OnReady(func(err error) {
+		ready <- err
+	})
+	lc.Append(fx.Hook{
+		OnStart: func(ctx context.Context) error {
+			if err := <-ready; err != nil {
+				return fmt.Errorf("aborting dagstore start; piecestore failed to start: %s", err)
+			}
+			return mountApi.Start(ctx)
+		},
+		OnStop: func(context.Context) error {
+			return nil
+		},
+	})
+
+	return mountApi, nil
+}
+
+// DAGStore constructs a DAG store using the supplied minerAPI, and the
+// user configuration. It returns both the DAGStore and the Wrapper suitable for
+// passing to markets.
+func DAGStore(lc fx.Lifecycle, r repo.LockedRepo, minerAPI mdagstore.MinerAPI) (*dagstore.DAGStore, *mdagstore.Wrapper, error) {
+	cfg, err := extractDAGStoreConfig(r)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	// fall back to default root directory if not explicitly set in the config.
+	if cfg.RootDir == "" {
+		cfg.RootDir = filepath.Join(r.Path(), DefaultDAGStoreDir)
+	}
+
+	v, ok := os.LookupEnv(EnvDAGStoreCopyConcurrency)
+	if ok {
+		concurrency, err := strconv.Atoi(v)
+		if err == nil {
+			cfg.MaxConcurrentReadyFetches = concurrency
+		}
+	}
+
+	dagst, w, err := mdagstore.NewDAGStore(cfg, minerAPI)
+	if err != nil {
+		return nil, nil, xerrors.Errorf("failed to create DAG store: %w", err)
+	}
+
+	lc.Append(fx.Hook{
+		OnStart: func(ctx context.Context) error {
+			return w.Start(ctx)
+		},
+		OnStop: func(context.Context) error {
+			return w.Close()
+		},
+	})
+
+	return dagst, w, nil
+}
+
+func extractDAGStoreConfig(r repo.LockedRepo) (config.DAGStoreConfig, error) {
+	cfg, err := r.Config()
+	if err != nil {
+		return config.DAGStoreConfig{}, xerrors.Errorf("could not load config: %w", err)
+	}
+	mcfg, ok := cfg.(*config.StorageMiner)
+	if !ok {
+		return config.DAGStoreConfig{}, xerrors.Errorf("config not expected type; expected config.StorageMiner, got: %T", cfg)
+	}
+	return mcfg.DAGStore, nil
+}