remove --halt-after-import flag from lotus daemon

CHANGELOG.md

@@ -1,12 +1,10 @@
 # Lotus changelog
 
-# 1.11.1-rc2 / 2021-08-01
+# 1.11.1-rc1 / 2021-07-27
 
-> 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)
+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)
 
 - github.com/filecoin-project/lotus:
   - Merge branch 'releases' into release/v1.11.1

api/api_storage.go

@@ -13,13 +13,12 @@ import (
 
 	"github.com/filecoin-project/go-address"
 	datatransfer "github.com/filecoin-project/go-data-transfer"
-	"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/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/lotus/chain/types"
 	"github.com/filecoin-project/lotus/extern/sector-storage/fsutil"
@@ -167,48 +166,6 @@ 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
@@ -379,33 +336,3 @@ 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 or struct types not implementing json.Marshaller", m.Name)
+							t.Error("methods can't return interfaces", m.Name)
 						}
 
 						switch typ.Kind() {

api/docgen/docgen.go

@@ -14,6 +14,7 @@ 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"
@@ -24,8 +25,6 @@ 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"
@@ -276,15 +275,6 @@ 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,16 +603,6 @@ 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"`
@@ -3579,61 +3569,6 @@ 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-m1.3.4"
+const BuildVersion = "1.11.1-rc1"
 
 func UserVersion() string {
 	if os.Getenv("LOTUS_VERSION_IGNORE_COMMIT") == "1" {

cli/util/api.go

@@ -2,7 +2,6 @@ package cliutil
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"net/http"
 	"net/url"
@@ -143,15 +142,6 @@ 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.Version(ctx)
+			_, err = api.ID(ctx)
 			if err != nil {
 				return err
 			}

cmd/lotus-miner/dagstore.go

@@ -1,209 +0,0 @@
-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,15 +5,13 @@ 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"
@@ -48,7 +46,6 @@ 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.GetMarketsAPI(cctx)
+		nodeApi, closer, err := lcli.GetStorageMinerAPI(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.GetMarketsAPI(cctx)
+		nodeApi, closer, err := lcli.GetStorageMinerAPI(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.GetMarketsAPI(cctx)
+		nodeApi, closer, err := lcli.GetStorageMinerAPI(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.GetMarketsAPI(cctx)
+		nodeApi, closer, err := lcli.GetStorageMinerAPI(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.GetMarketsAPI(cctx)
+		nodeApi, closer, err := lcli.GetStorageMinerAPI(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.GetMarketsAPI(cctx)
+		nodeApi, closer, err := lcli.GetStorageMinerAPI(cctx)
 		if err != nil {
 			return err
 		}

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

@@ -18,12 +18,6 @@
   * [ComputeProof](#ComputeProof)
 * [Create](#Create)
   * [CreateBackup](#CreateBackup)
-* [Dagstore](#Dagstore)
-  * [DagstoreGC](#DagstoreGC)
-  * [DagstoreInitializeAll](#DagstoreInitializeAll)
-  * [DagstoreInitializeShard](#DagstoreInitializeShard)
-  * [DagstoreListShards](#DagstoreListShards)
-  * [DagstoreRecoverShard](#DagstoreRecoverShard)
 * [Deals](#Deals)
   * [DealsConsiderOfflineRetrievalDeals](#DealsConsiderOfflineRetrievalDeals)
   * [DealsConsiderOfflineStorageDeals](#DealsConsiderOfflineStorageDeals)
@@ -351,113 +345,6 @@ 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
 
 
@@ -822,7 +709,7 @@ Response:
     "ID": 3
   },
   "SectorNumber": 9,
-  "InboundCAR": "string value"
+  "CARv2FilePath": "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-m1.3.4
+   1.11.1-rc1
 
 COMMANDS:
    init     Initialize a lotus miner repo
@@ -29,7 +29,6 @@ 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:
@@ -1001,94 +1000,6 @@ 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-m1.3.4
+   1.11.1-rc1
 
 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-m1.3.4
+   1.11.1-rc1
 
 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.4.2
+	github.com/filecoin-project/dagstore v0.3.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.3
+	github.com/filecoin-project/go-data-transfer v1.7.2
 	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.20210806151421-967d8717c917
+	github.com/filecoin-project/go-fil-markets v1.6.3-0.20210729173742-a44f98a4f8c1
 	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.6.0
+	github.com/golang/mock v1.5.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.8
+	github.com/ipfs/go-graphsync v0.6.6
 	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.20210803143113-d0e44a62f0a3
+	github.com/ipld/go-car/v2 v2.0.0-beta1.0.20210721090610-5a9d1b217d25
 	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,8 +256,9 @@ 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.4.2 h1:Ae2+O1DhKCI1JbOZCBkqUksKYofdbRbjkS7OF0A6Jw0=
-github.com/filecoin-project/dagstore v0.4.2/go.mod h1:WY5OoLfnwISCk6eASSF927KKPqLPIlTwmG1qHpA08KY=
+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/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=
@@ -278,8 +279,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.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-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-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=
@@ -289,8 +290,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.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-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-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=
@@ -434,8 +435,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.6.0 h1:ErTB+efbowRARo13NNdxyJji2egdxLGQhRaY+DUumQc=
-github.com/golang/mock v1.6.0/go.mod h1:p6yTPP+5HYm5mzsMV8JkE6ZKdX+/wYM6Hr+LicevLPs=
+github.com/golang/mock v1.5.0 h1:jlYHihg//f7RRwuPfptm04yp4s7O6Kw8EZiVYIGcH0g=
+github.com/golang/mock v1.5.0/go.mod h1:CWnOUgYIOo4TcNZ0wHX3YZCqsaM1I1Jvs6v3mP3KVu8=
 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=
@@ -639,8 +640,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.8 h1:mgyPdBDPcgL8ujO132grQjP3rfQv+KN/riQzbmTVgg4=
-github.com/ipfs/go-graphsync v0.6.8/go.mod h1:GdHT8JeuIZ0R4lSjFR16Oe4zPi5dXwKi9zR9ADVlcdk=
+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-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=
@@ -759,9 +760,8 @@ 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,7 +1989,6 @@ 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

@@ -1,33 +0,0 @@
-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/lotusaccessor.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"os"
+	"strconv"
 
 	"github.com/filecoin-project/dagstore/throttle"
 	"github.com/ipfs/go-cid"
@@ -14,84 +16,49 @@ import (
 	"github.com/filecoin-project/go-fil-markets/shared"
 )
 
-type MinerAPI interface {
+// 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 {
 	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 minerAPI struct {
+type lotusAccessor struct {
 	pieceStore piecestore.PieceStore
 	rm         retrievalmarket.RetrievalProviderNode
 	throttle   throttle.Throttler
 	readyMgr   *shared.ReadyManager
 }
 
-var _ MinerAPI = (*minerAPI)(nil)
+var _ LotusAccessor = (*lotusAccessor)(nil)
 
-func NewMinerAPI(store piecestore.PieceStore, rm retrievalmarket.RetrievalProviderNode, concurrency int) MinerAPI {
-	return &minerAPI{
+func NewLotusAccessor(store piecestore.PieceStore, rm retrievalmarket.RetrievalProviderNode) LotusAccessor {
+	return &lotusAccessor{
 		pieceStore: store,
 		rm:         rm,
-		throttle:   throttle.Fixed(concurrency),
+		throttle:   throttle.Fixed(MaxConcurrentStorageCalls),
 		readyMgr:   shared.NewReadyManager(),
 	}
 }
 
-func (m *minerAPI) Start(_ context.Context) error {
+func (m *lotusAccessor) Start(_ context.Context) error {
 	return m.readyMgr.FireReady(nil)
 }
 
-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) {
+func (m *lotusAccessor) FetchUnsealedPiece(ctx context.Context, pieceCid cid.Cid) (io.ReadCloser, error) {
 	err := m.readyMgr.AwaitReady()
 	if err != nil {
 		return nil, err
@@ -163,7 +130,7 @@ func (m *minerAPI) FetchUnsealedPiece(ctx context.Context, pieceCid cid.Cid) (io
 	return nil, lastErr
 }
 
-func (m *minerAPI) GetUnpaddedCARSize(ctx context.Context, pieceCid cid.Cid) (uint64, error) {
+func (m *lotusAccessor) GetUnpaddedCARSize(ctx context.Context, pieceCid cid.Cid) (uint64, error) {
 	err := m.readyMgr.AwaitReady()
 	if err != nil {
 		return 0, err

markets/dagstore/lotusaccessor_test.go

@@ -45,9 +45,7 @@ func TestLotusAccessorFetchUnsealedPiece(t *testing.T) {
 		name        string
 		deals       []abi.SectorNumber
 		fetchedData string
-		isUnsealed  bool
-
-		expectErr bool
+		expectErr   bool
 	}{{
 		// Expect error if there is no deal info for piece CID
 		name:      "no deals",
@@ -58,13 +56,11 @@ 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 {
@@ -74,7 +70,7 @@ func TestLotusAccessorFetchUnsealedPiece(t *testing.T) {
 			rpn := &mockRPN{
 				sectors: mockData,
 			}
-			api := NewMinerAPI(ps, rpn, 100)
+			api := NewLotusAccessor(ps, rpn)
 			require.NoError(t, api.Start(ctx))
 
 			// Add deals to piece store
@@ -99,10 +95,6 @@ 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)
 		})
 	}
 }
@@ -114,7 +106,7 @@ func TestLotusAccessorGetUnpaddedCARSize(t *testing.T) {
 
 	ps := getPieceStore(t)
 	rpn := &mockRPN{}
-	api := NewMinerAPI(ps, rpn, 100)
+	api := NewLotusAccessor(ps, rpn)
 	require.NoError(t, api.Start(ctx))
 
 	// Add a deal with data Length 10
@@ -131,6 +123,8 @@ func TestLotusAccessorGetUnpaddedCARSize(t *testing.T) {
 }
 
 func TestThrottle(t *testing.T) {
+	MaxConcurrentStorageCalls = 3
+
 	ctx := context.Background()
 	cid1, err := cid.Parse("bafkqaaa")
 	require.NoError(t, err)
@@ -141,7 +135,7 @@ func TestThrottle(t *testing.T) {
 			unsealedSectorID: "foo",
 		},
 	}
-	api := NewMinerAPI(ps, rpn, 3)
+	api := NewLotusAccessor(ps, rpn)
 	require.NoError(t, api.Start(ctx))
 
 	// Add a deal with data Length 10
@@ -168,7 +162,7 @@ func TestThrottle(t *testing.T) {
 	}
 
 	time.Sleep(500 * time.Millisecond)
-	require.EqualValues(t, 3, atomic.LoadInt32(&rpn.calls)) // throttled
+	require.EqualValues(t, MaxConcurrentStorageCalls, atomic.LoadInt32(&rpn.calls)) // throttled
 
 	// allow to proceed.
 	rpn.lk.Unlock()
@@ -185,14 +179,8 @@ 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: lotusaccessor.go
+// Source: markets/dagstore/lotusaccessor.go
 
 // Package mock_dagstore is a generated GoMock package.
 package mock_dagstore
@@ -31,6 +31,10 @@ 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
@@ -65,32 +69,3 @@ 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,29 +15,28 @@ const lotusScheme = "lotus"
 
 var _ mount.Mount = (*LotusMount)(nil)
 
-// 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.
-//
-// 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 mountTemplate(api MinerAPI) *LotusMount {
-	return &LotusMount{API: api}
-}
-
-// LotusMount is a DAGStore mount implementation that fetches deal data
-// from a PieceCID.
+// 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      MinerAPI
+	Api      LotusAccessor
 	PieceCid cid.Cid
 }
 
-func NewLotusMount(pieceCid cid.Cid, api MinerAPI) (*LotusMount, error) {
+// NewLotusMountTemplate is called when registering a mount with
+// the DAG store registry.
+//
+// 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 NewLotusMount(pieceCid cid.Cid, api LotusAccessor) (*LotusMount, error) {
 	return &LotusMount{
 		PieceCid: pieceCid,
-		API:      api,
+		Api:      api,
 	}, nil
 }
 
@@ -52,12 +51,13 @@ 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,20 +78,15 @@ 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,9 +27,6 @@ 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)
@@ -55,7 +52,7 @@ func TestLotusMount(t *testing.T) {
 	// serialize url then deserialize from mount template -> should get back
 	// the same mount
 	url := mnt.Serialize()
-	mnt2 := mountTemplate(mockLotusMountAPI)
+	mnt2 := NewLotusMountTemplate(mockLotusMountAPI)
 	err = mnt2.Deserialize(url)
 	require.NoError(t, err)
 
@@ -69,7 +66,7 @@ func TestLotusMount(t *testing.T) {
 }
 
 func TestLotusMountDeserialize(t *testing.T) {
-	api := &minerAPI{}
+	api := &lotusAccessor{}
 
 	bgen := blocksutil.NewBlockGenerator()
 	cid := bgen.Next().Cid()
@@ -79,12 +76,12 @@ func TestLotusMountDeserialize(t *testing.T) {
 	u, err := url.Parse(us)
 	require.NoError(t, err)
 
-	mnt := mountTemplate(api)
+	mnt := NewLotusMountTemplate(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"
@@ -111,16 +108,14 @@ func TestLotusMountRegistration(t *testing.T) {
 
 	mockLotusMountAPI := mock_dagstore.NewMockLotusAccessor(mockCtrl)
 	registry := mount.NewRegistry()
-	err = registry.Register(lotusScheme, mountTemplate(mockLotusMountAPI))
+	err = registry.Register(lotusScheme, NewLotusMountTemplate(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

@@ -0,0 +1,33 @@
+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,202 +3,174 @@ package dagstore
 import (
 	"context"
 	"errors"
-	"math"
-	"os"
-	"path/filepath"
+	"io"
 	"sync"
 	"time"
 
+	"github.com/filecoin-project/dagstore/index"
 	"github.com/ipfs/go-cid"
 	ds "github.com/ipfs/go-datastore"
-	levelds "github.com/ipfs/go-ds-leveldb"
-	measure "github.com/ipfs/go-ds-measure"
+	bstore "github.com/ipfs/go-ipfs-blockstore"
 	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/storagemarket"
-	"github.com/filecoin-project/go-fil-markets/storagemarket/impl/providerstates"
-	"github.com/filecoin-project/go-fil-markets/stores"
+	"github.com/filecoin-project/go-fil-markets/carstore"
+	"github.com/filecoin-project/go-fil-markets/shared"
 )
 
-const (
-	maxRecoverAttempts = 1
-	shardRegMarker     = ".shard-registration-complete"
-)
+const maxRecoverAttempts = 1
 
-var log = logging.Logger("dagstore")
+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
+}
 
 type Wrapper struct {
 	ctx          context.Context
 	cancel       context.CancelFunc
 	backgroundWg sync.WaitGroup
 
-	cfg        config.DAGStoreConfig
-	dagst      dagstore.Interface
-	minerAPI   MinerAPI
+	dagStore   DAGStore
+	mountApi   LotusAccessor
 	failureCh  chan dagstore.ShardResult
 	traceCh    chan dagstore.Trace
 	gcInterval time.Duration
 }
 
-var _ stores.DAGStoreWrapper = (*Wrapper)(nil)
+var _ shared.DagStoreWrapper = (*Wrapper)(nil)
 
-func NewDAGStore(cfg config.DAGStoreConfig, mountApi MinerAPI) (*dagstore.DAGStore, *Wrapper, error) {
+func NewDagStoreWrapper(cfg MarketDAGStoreConfig, mountApi LotusAccessor) (*Wrapper, error) {
 	// construct the DAG Store.
 	registry := mount.NewRegistry()
-	if err := registry.Register(lotusScheme, mountTemplate(mountApi)); err != nil {
-		return nil, nil, xerrors.Errorf("failed to create registry: %w", err)
+	if err := registry.Register(lotusScheme, NewLotusMountTemplate(mountApi)); err != nil {
+		return 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)
 
-	var (
-		transientsDir = filepath.Join(cfg.RootDir, "transients")
-		datastoreDir  = filepath.Join(cfg.RootDir, "datastore")
-		indexDir      = filepath.Join(cfg.RootDir, "index")
-	)
-
-	dstore, err := newDatastore(datastoreDir)
+	irepo, err := index.NewFSRepo(cfg.IndexDir)
 	if err != nil {
-		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")
+		return nil, xerrors.Errorf("failed to initialise dagstore index repo")
 	}
 
 	dcfg := dagstore.Config{
-		TransientsDir: transientsDir,
+		TransientsDir: cfg.TransientsDir,
 		IndexRepo:     irepo,
-		Datastore:     dstore,
+		Datastore:     cfg.Datastore,
 		MountRegistry: registry,
 		FailureCh:     failureCh,
 		TraceCh:       traceCh,
 		// not limiting fetches globally, as the Lotus mount does
 		// conditional throttling.
-		MaxConcurrentIndex:        cfg.MaxConcurrentIndex,
-		MaxConcurrentReadyFetches: cfg.MaxConcurrentReadyFetches,
-		RecoverOnStart:            dagstore.RecoverOnAcquire,
+		MaxConcurrentIndex:  cfg.MaxConcurrentIndex,
+		MaxConcurrentCopies: cfg.MaxConcurrentCopies,
+		RecoverOnStart:      dagstore.RecoverOnAcquire,
 	}
-
-	dagst, err := dagstore.NewDAGStore(dcfg)
+	dagStore, err := dagstore.NewDAGStore(dcfg)
 	if err != nil {
-		return nil, nil, xerrors.Errorf("failed to create DAG store: %w", err)
+		return nil, xerrors.Errorf("failed to create DAG store: %w", err)
 	}
 
-	w := &Wrapper{
-		cfg:        cfg,
-		dagst:      dagst,
-		minerAPI:   mountApi,
+	return &Wrapper{
+		dagStore:   dagStore,
+		mountApi:   mountApi,
 		failureCh:  failureCh,
 		traceCh:    traceCh,
-		gcInterval: time.Duration(cfg.GCInterval),
-	}
-
-	return dagst, w, nil
+		gcInterval: cfg.GCInterval,
+	}, nil
 }
 
-// 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)
-	}
-
-	// 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)
+func (ds *Wrapper) Start(ctx context.Context) error {
+	ds.ctx, ds.cancel = context.WithCancel(ctx)
 
 	// Run a go-routine to do DagStore GC.
-	w.backgroundWg.Add(1)
-	go w.gcLoop()
+	ds.backgroundWg.Add(1)
+	go ds.dagStoreGCLoop()
 
 	// run a go-routine to read the trace for debugging.
-	w.backgroundWg.Add(1)
-	go w.traceLoop()
+	ds.backgroundWg.Add(1)
+	go ds.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)
+	if dss, ok := ds.dagStore.(*dagstore.DAGStore); ok {
+		ds.backgroundWg.Add(1)
+		go dagstore.RecoverImmediately(ds.ctx, dss, ds.failureCh, maxRecoverAttempts, ds.backgroundWg.Done)
 	}
 
-	return w.dagst.Start(ctx)
+	return ds.dagStore.Start(ctx)
 }
 
-func (w *Wrapper) traceLoop() {
-	defer w.backgroundWg.Done()
+func (ds *Wrapper) traceLoop() {
+	defer ds.backgroundWg.Done()
 
-	for w.ctx.Err() == nil {
+	for ds.ctx.Err() == nil {
 		select {
 		// Log trace events from the DAG store
-		case tr := <-w.traceCh:
+		case tr := <-ds.traceCh:
 			log.Debugw("trace",
 				"shard-key", tr.Key.String(),
 				"op-type", tr.Op.String(),
 				"after", tr.After.String())
 
-		case <-w.ctx.Done():
+		case <-ds.ctx.Done():
 			return
 		}
 	}
 }
 
-func (w *Wrapper) gcLoop() {
-	defer w.backgroundWg.Done()
+func (ds *Wrapper) dagStoreGCLoop() {
+	defer ds.backgroundWg.Done()
 
-	ticker := time.NewTicker(w.gcInterval)
-	defer ticker.Stop()
+	gcTicker := time.NewTicker(ds.gcInterval)
+	defer gcTicker.Stop()
 
-	for w.ctx.Err() == nil {
+	for ds.ctx.Err() == nil {
 		select {
 		// GC the DAG store on every tick
-		case <-ticker.C:
-			_, _ = w.dagst.GC(w.ctx)
+		case <-gcTicker.C:
+			_, _ = ds.dagStore.GC(ds.ctx)
 
 		// Exit when the DAG store wrapper is shutdown
-		case <-w.ctx.Done():
+		case <-ds.ctx.Done():
 			return
 		}
 	}
 }
 
-func (w *Wrapper) LoadShard(ctx context.Context, pieceCid cid.Cid) (stores.ClosableBlockstore, error) {
+func (ds *Wrapper) LoadShard(ctx context.Context, pieceCid cid.Cid) (carstore.ClosableBlockstore, error) {
 	log.Debugf("acquiring shard for piece CID %s", pieceCid)
-
 	key := shard.KeyFromCID(pieceCid)
 	resch := make(chan dagstore.ShardResult, 1)
-	err := w.dagst.AcquireShard(ctx, key, resch, dagstore.AcquireOpts{})
+	err := ds.dagStore.AcquireShard(ctx, key, resch, dagstore.AcquireOpts{})
 	log.Debugf("sent message to acquire shard for piece CID %s", pieceCid)
 
 	if err != nil {
@@ -213,13 +185,13 @@ func (w *Wrapper) LoadShard(ctx context.Context, pieceCid cid.Cid) (stores.Closa
 		// we already have a transient file. However, we don't have it here
 		// and therefore we pass an empty file path.
 		carPath := ""
-		if err := stores.RegisterShardSync(ctx, w, pieceCid, carPath, false); err != nil {
+		if err := shared.RegisterShardSync(ctx, ds, 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 := w.dagst.AcquireShard(ctx, key, resch, dagstore.AcquireOpts{}); err != nil {
+		if err := ds.dagStore.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)
 		}
 	}
@@ -244,13 +216,13 @@ func (w *Wrapper) LoadShard(ctx context.Context, pieceCid cid.Cid) (stores.Closa
 	}
 
 	log.Debugf("successfully loaded blockstore for piece CID %s", pieceCid)
-	return &Blockstore{ReadBlockstore: bs, Closer: res.Accessor}, nil
+	return &closableBlockstore{Blockstore: NewReadOnlyBlockstore(bs), Closer: res.Accessor}, nil
 }
 
-func (w *Wrapper) RegisterShard(ctx context.Context, pieceCid cid.Cid, carPath string, eagerInit bool, resch chan dagstore.ShardResult) error {
+func (ds *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, w.minerAPI)
+	mt, err := NewLotusMount(pieceCid, ds.mountApi)
 	if err != nil {
 		return xerrors.Errorf("failed to create lotus mount for piece CID %s: %w", pieceCid, err)
 	}
@@ -260,7 +232,7 @@ func (w *Wrapper) RegisterShard(ctx context.Context, pieceCid cid.Cid, carPath s
 		ExistingTransient:  carPath,
 		LazyInitialization: !eagerInit,
 	}
-	err = w.dagst.RegisterShard(ctx, key, mt, resch, opts)
+	err = ds.dagStore.RegisterShard(ctx, key, mt, resch, opts)
 	if err != nil {
 		return xerrors.Errorf("failed to schedule register shard for piece CID %s: %w", pieceCid, err)
 	}
@@ -269,153 +241,21 @@ func (w *Wrapper) RegisterShard(ctx context.Context, pieceCid cid.Cid, carPath s
 	return nil
 }
 
-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 {
+func (ds *Wrapper) Close() error {
 	// Cancel the context
-	w.cancel()
+	ds.cancel()
 
 	// Close the DAG store
 	log.Info("will close the dagstore")
-	if err := w.dagst.Close(); err != nil {
-		return xerrors.Errorf("failed to close dagstore: %w", err)
+	if err := ds.dagStore.Close(); err != nil {
+		return xerrors.Errorf("failed to close DAG store: %w", err)
 	}
 	log.Info("dagstore closed")
 
 	// Wait for the background go routine to exit
-	log.Info("waiting for dagstore background wrapper goroutines to exit")
-	w.backgroundWg.Wait()
-	log.Info("exited dagstore background wrapper goroutines")
+	log.Info("waiting for dagstore background wrapper routines to exist")
+	ds.backgroundWg.Wait()
+	log.Info("exited dagstore background warpper routines")
 
 	return nil
 }

markets/dagstore/wrapper_migration_test.go

@@ -1,110 +0,0 @@
-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,8 +10,6 @@ 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"
@@ -28,14 +26,13 @@ func TestWrapperAcquireRecovery(t *testing.T) {
 	require.NoError(t, err)
 
 	// Create a DAG store wrapper
-	dagst, w, err := NewDAGStore(config.DAGStoreConfig{
-		RootDir:    t.TempDir(),
-		GCInterval: config.Duration(1 * time.Millisecond),
+	w, err := NewDagStoreWrapper(MarketDAGStoreConfig{
+		TransientsDir: t.TempDir(),
+		IndexDir:      t.TempDir(),
+		GCInterval:    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)
@@ -48,7 +45,7 @@ func TestWrapperAcquireRecovery(t *testing.T) {
 		},
 		register: make(chan shard.Key, 1),
 	}
-	w.dagst = mock
+	w.dagStore = mock
 
 	mybs, err := w.LoadShard(ctx, pieceCid)
 	require.NoError(t, err)
@@ -79,25 +76,23 @@ func TestWrapperBackground(t *testing.T) {
 	ctx := context.Background()
 
 	// Create a DAG store wrapper
-	dagst, w, err := NewDAGStore(config.DAGStoreConfig{
-		RootDir:    t.TempDir(),
-		GCInterval: config.Duration(1 * time.Millisecond),
+	w, err := NewDagStoreWrapper(MarketDAGStoreConfig{
+		TransientsDir: t.TempDir(),
+		IndexDir:      t.TempDir(),
+		GCInterval:    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.dagst = mock
+	w.dagStore = mock
 
 	// Start up the wrapper
-	err = w.Start(ctx)
-	require.NoError(t, err)
+	w.Start(ctx)
 
 	// Expect GC to be called automatically
 	tctx, cancel := context.WithTimeout(ctx, time.Second)
@@ -132,22 +127,6 @@ 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}
@@ -199,10 +178,6 @@ 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,7 +66,6 @@ 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.MinerAPI), modules.NewMinerAPI),
-			Override(DAGStoreKey, modules.DAGStore),
+			Override(new(dagstore.LotusAccessor), modules.NewLotusAccessor),
+			Override(new(*dagstore.Wrapper), modules.DagStoreWrapper),
 
 			// Markets (retrieval)
 			Override(new(retrievalmarket.RetrievalProviderNode), retrievaladapter.NewRetrievalProviderNode),

node/config/def.go

@@ -188,12 +188,6 @@ 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,46 +125,6 @@ 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",
@@ -793,12 +753,6 @@ Default is 20 (about once a week).`,
 			Name: "Addresses",
 			Type: "MinerAddressConfig",
 
-			Comment: ``,
-		},
-		{
-			Name: "DAGStore",
-			Type: "DAGStoreConfig",
-
 			Comment: ``,
 		},
 	},

node/config/types.go

@@ -50,42 +50,6 @@ 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,7 +10,8 @@ import (
 	"sort"
 	"time"
 
-	"github.com/ipld/go-car"
+	"github.com/filecoin-project/go-fil-markets/filestorecaradapter"
+	"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
 	carv2 "github.com/ipld/go-car/v2"
 	"github.com/ipld/go-car/v2/blockstore"
 
@@ -25,7 +26,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"
@@ -46,7 +47,6 @@ 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,7 +54,6 @@ 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"
@@ -189,17 +188,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,
-			IndexedCAR:    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,
+			FilestoreCARv2FilePath: CARV2FilePath,
 		})
 
 		if err != nil {
@@ -520,7 +519,7 @@ func (a *API) ClientImport(ctx context.Context, ref api.FileRef) (res *api.Impor
 			}
 		}()
 
-		root, err = a.doImport(ctx, ref.Path, carFile)
+		root, err = a.importNormalFileToFilestoreCARv2(ctx, id, ref.Path, carFile)
 		if err != nil {
 			return nil, xerrors.Errorf("failed to import normal file to CARv2: %w", err)
 		}
@@ -1032,21 +1031,21 @@ func (w *lenWriter) Write(p []byte) (n int, err error) {
 }
 
 func (a *API) ClientDealSize(ctx context.Context, root cid.Cid) (api.DataSize, error) {
-	path, err := a.imgr().FilestoreCARV2FilePathFor(root)
+	fc, err := a.imgr().FilestoreCARV2FilePathFor(root)
 	if err != nil {
 		return api.DataSize{}, xerrors.Errorf("failed to find CARv2 file for root: %w", err)
 	}
-	if len(path) == 0 {
+	if len(fc) == 0 {
 		return api.DataSize{}, xerrors.New("no CARv2 file for root")
 	}
 
-	fs, err := stores.ReadOnlyFilestore(path)
+	rdOnly, err := filestorecaradapter.NewReadOnlyFileStore(fc)
 	if err != nil {
-		return api.DataSize{}, xerrors.Errorf("failed to open filestore from carv2 in path %s: %w", path, err)
+		return api.DataSize{}, xerrors.Errorf("failed to open read only filestore: %w", err)
 	}
-	defer fs.Close() //nolint:errcheck
+	defer rdOnly.Close() //nolint:errcheck
 
-	dag := merkledag.NewDAGService(blockservice.New(fs, offline.Exchange(fs)))
+	dag := merkledag.NewDAGService(blockservice.New(rdOnly, offline.Exchange(rdOnly)))
 	w := lenWriter(0)
 
 	err = car.WriteCar(ctx, dag, []cid.Cid{root}, &w)
@@ -1063,21 +1062,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) {
-	path, err := a.imgr().FilestoreCARV2FilePathFor(root)
+	fc, err := a.imgr().FilestoreCARV2FilePathFor(root)
 	if err != nil {
 		return api.DataCIDSize{}, xerrors.Errorf("failed to find CARv2 file for root: %w", err)
 	}
-	if len(path) == 0 {
+	if len(fc) == 0 {
 		return api.DataCIDSize{}, xerrors.New("no CARv2 file for root")
 	}
 
-	fs, err := stores.ReadOnlyFilestore(path)
+	rdOnly, err := filestorecaradapter.NewReadOnlyFileStore(fc)
 	if err != nil {
-		return api.DataCIDSize{}, xerrors.Errorf("failed to open filestore from carv2 in path %s: %w", path, err)
+		return api.DataCIDSize{}, xerrors.Errorf("failed to open read only blockstore: %w", err)
 	}
-	defer fs.Close() //nolint:errcheck
+	defer rdOnly.Close() //nolint:errcheck
 
-	dag := merkledag.NewDAGService(blockservice.New(fs, offline.Exchange(fs)))
+	dag := merkledag.NewDAGService(blockservice.New(rdOnly, offline.Exchange(rdOnly)))
 	w := &writer.Writer{}
 	bw := bufio.NewWriterSize(w, int(writer.CommPBuf))
 
@@ -1096,20 +1095,22 @@ 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())
-	tmp, err := a.imgr().NewTempFile(id)
+	tmpCARv2File, err := a.imgr().NewTempFile(id)
 	if err != nil {
 		return xerrors.Errorf("failed to create temp file: %w", err)
 	}
-	defer os.Remove(tmp) //nolint:errcheck
+	defer os.Remove(tmpCARv2File) //nolint:errcheck
 
-	root, err := a.doImport(ctx, ref.Path, tmp)
+	root, err := a.importNormalFileToFilestoreCARv2(ctx, id, ref.Path, tmpCARv2File)
 	if err != nil {
 		return xerrors.Errorf("failed to import normal file to CARv2")
 	}
 
-	fs, err := stores.ReadOnlyFilestore(tmp)
+	// 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)
 	if err != nil {
-		return xerrors.Errorf("failed to open filestore from carv2 in path %s: %w", tmp, err)
+		return xerrors.Errorf("failed to open read only CARv2 blockstore: %w", err)
 	}
 	defer fs.Close() //nolint:errcheck
 

node/impl/client/import.go

@@ -2,13 +2,18 @@ package client
 
 import (
 	"context"
-	"io/ioutil"
 	"os"
 
-	"github.com/filecoin-project/go-fil-markets/stores"
+	"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/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"
@@ -16,83 +21,56 @@ 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"
 )
 
-// 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.
-	//
-	// 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)
-	}
+// 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.
+	//
+	// 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)
 	bsvc := blockservice.New(fstore, offline.Exchange(fstore))
-	dags := merkledag.NewDAGService(bsvc)
+	defer bsvc.Close() //nolint:errcheck
 
-	root, err := buildUnixFS(ctx, src, dags)
+	// ---- 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))
 	if err != nil {
-		_ = fstore.Close()
-		return cid.Undef, xerrors.Errorf("failed to import file to store to compute root: %w", err)
+		return cid.Undef, xerrors.Errorf("failed to import file to store: %w", err)
 	}
 
-	if err := fstore.Close(); err != nil {
-		return cid.Undef, xerrors.Errorf("failed to finalize car filestore: %w", err)
-	}
-
-	// 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))
+	//------ 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)
+		return cid.Undef, xerrors.Errorf("failed to create a CARv2 read-write blockstore: %w", err)
 	}
+	defer fsb.Close() //nolint:errcheck
 
-	bsvc = blockservice.New(bs, offline.Exchange(bs))
-	dags = merkledag.NewDAGService(bsvc)
-	finalRoot, err := buildUnixFS(ctx, src, dags)
+	bsvc = blockservice.New(fsb, offline.Exchange(fsb))
+	root2, err := importNormalFileToUnixfsDAG(ctx, inputFilePath, merkledag.NewDAGService(bsvc))
 	if err != nil {
-		_ = bs.Close()
-		return cid.Undef, xerrors.Errorf("failed to create UnixFS DAG with carv2 blockstore: %w", err)
+		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 {
+	if root != root2 {
 		return cid.Undef, xerrors.New("roots do not match")
 	}
 
 	return root, nil
 }
 
-// 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)
+// 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)
 	if err != nil {
 		return cid.Undef, xerrors.Errorf("failed to open input file: %w", err)
 	}
@@ -103,7 +81,7 @@ func buildUnixFS(ctx context.Context, src string, dag ipld.DAGService) (cid.Cid,
 		return cid.Undef, xerrors.Errorf("failed to stat file :%w", err)
 	}
 
-	file, err := files2.NewReaderPathFile(src, f, stat)
+	file, err := files2.NewReaderPathFile(inputFilePath, 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,11 +4,12 @@ import (
 	"context"
 	"io"
 	"io/ioutil"
+	"math/rand"
 	"os"
 	"strings"
 	"testing"
 
-	"github.com/filecoin-project/go-fil-markets/stores"
+	"github.com/filecoin-project/go-fil-markets/filestorecaradapter"
 	"github.com/filecoin-project/lotus/node/repo/importmgr"
 	"github.com/ipfs/go-blockservice"
 	"github.com/ipfs/go-cid"
@@ -21,50 +22,39 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-// This test uses a full "dense" CARv2, and not a filestore (positional mapping).
-func TestRoundtripUnixFS_Dense(t *testing.T) {
+func TestImportNormalFileToUnixfsDAG(t *testing.T) {
 	ctx := context.Background()
-
-	inputPath, inputContents := genInputFile(t)
+	inputPath, inputContents := genNormalInputFile(t)
 	defer os.Remove(inputPath) //nolint:errcheck
-
-	carv2File := newTmpFile(t)
+	carv2File := genTmpFile(t)
 	defer os.Remove(carv2File) //nolint:errcheck
 
-	// 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))
+	// 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))
 	require.NoError(t, err)
-
-	bsvc := blockservice.New(path, offline.Exchange(path))
-	dags := merkledag.NewDAGService(bsvc)
-
-	root, err := buildUnixFS(ctx, inputPath, dags)
+	bsvc := blockservice.New(tempCARv2Store, offline.Exchange(tempCARv2Store))
+	root, err := importNormalFileToUnixfsDAG(ctx, inputPath, merkledag.NewDAGService(bsvc))
 	require.NoError(t, err)
 	require.NotEqual(t, cid.Undef, root)
-	require.NoError(t, path.Finalize())
+	require.NoError(t, tempCARv2Store.Finalize())
 
-	// reconstruct the file.
-	readOnly, err := blockstore.OpenReadOnly(carv2File,
-		carv2.ZeroLengthSectionAsEOF(true),
-		blockstore.UseWholeCIDs(true))
+	// 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))
 	require.NoError(t, err)
 	defer readOnly.Close() //nolint:errcheck
+	dag := merkledag.NewDAGService(blockservice.New(readOnly, offline.Exchange(readOnly)))
 
-	dags = merkledag.NewDAGService(blockservice.New(readOnly, offline.Exchange(readOnly)))
-
-	nd, err := dags.Get(ctx, root)
+	nd, err := dag.Get(ctx, root)
+	require.NoError(t, err)
+	file, err := unixfile.NewUnixfsFile(ctx, dag, nd)
 	require.NoError(t, err)
 
-	file, err := unixfile.NewUnixfsFile(ctx, dags, nd)
-	require.NoError(t, err)
-
-	tmpOutput := newTmpFile(t)
+	tmpOutput := genTmpFile(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)
@@ -73,36 +63,35 @@ func TestRoundtripUnixFS_Dense(t *testing.T) {
 	require.Equal(t, inputContents, bz2)
 }
 
-func TestRoundtripUnixFS_Filestore(t *testing.T) {
+func TestImportNormalFileToCARv2(t *testing.T) {
 	ctx := context.Background()
 	a := &API{
 		Imports: &importmgr.Mgr{},
 	}
+	importID := importmgr.ImportID(rand.Uint64())
 
-	inputFilePath, inputContents := genInputFile(t)
+	inputFilePath, inputContents := genNormalInputFile(t)
 	defer os.Remove(inputFilePath) //nolint:errcheck
 
-	path := newTmpFile(t)
-	defer os.Remove(path) //nolint:errcheck
+	outputCARv2 := genTmpFile(t)
+	defer os.Remove(outputCARv2) //nolint:errcheck
 
-	root, err := a.doImport(ctx, inputFilePath, path)
+	root, err := a.importNormalFileToFilestoreCARv2(ctx, importID, inputFilePath, outputCARv2)
 	require.NoError(t, err)
 	require.NotEqual(t, cid.Undef, root)
 
 	// convert the CARv2 to a normal file again and ensure the contents match
-	fs, err := stores.ReadOnlyFilestore(path)
+	readOnly, err := filestorecaradapter.NewReadOnlyFileStore(outputCARv2)
 	require.NoError(t, err)
-	defer fs.Close() //nolint:errcheck
+	defer readOnly.Close() //nolint:errcheck
+	dag := merkledag.NewDAGService(blockservice.New(readOnly, offline.Exchange(readOnly)))
 
-	dags := merkledag.NewDAGService(blockservice.New(fs, offline.Exchange(fs)))
-
-	nd, err := dags.Get(ctx, root)
+	nd, err := dag.Get(ctx, root)
+	require.NoError(t, err)
+	file, err := unixfile.NewUnixfsFile(ctx, dag, nd)
 	require.NoError(t, err)
 
-	file, err := unixfile.NewUnixfsFile(ctx, dags, nd)
-	require.NoError(t, err)
-
-	tmpOutput := newTmpFile(t)
+	tmpOutput := genTmpFile(t)
 	defer os.Remove(tmpOutput) //nolint:errcheck
 	require.NoError(t, files.WriteTo(file, tmpOutput))
 
@@ -115,14 +104,14 @@ func TestRoundtripUnixFS_Filestore(t *testing.T) {
 	require.Equal(t, inputContents, bz2)
 }
 
-func newTmpFile(t *testing.T) string {
+func genTmpFile(t *testing.T) string {
 	f, err := os.CreateTemp("", "")
 	require.NoError(t, err)
 	require.NoError(t, f.Close())
 	return f.Name()
 }
 
-func genInputFile(t *testing.T) (filepath string, contents []byte) {
+func genNormalInputFile(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,20 +3,16 @@ 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"
@@ -24,8 +20,6 @@ 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"
@@ -40,8 +34,6 @@ 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"
@@ -50,6 +42,7 @@ 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 {
@@ -72,7 +65,6 @@ 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"`
@@ -106,8 +98,6 @@ 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 {
@@ -555,225 +545,6 @@ 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)
 }
@@ -937,3 +708,5 @@ 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(), DefaultDAGStoreDir, "retrieval-cars")
+	carsPath := filepath.Join(r.Path(), dagStore, "retrieval-cars")
 
 	if err := os.MkdirAll(carsPath, 0755); err != nil {
 		return nil, xerrors.Errorf("failed to create dir")

node/modules/storageminer.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"net/http"
 	"os"
+	"path"
 	"path/filepath"
 	"strings"
 	"time"
@@ -15,28 +16,26 @@ 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"
@@ -47,6 +46,11 @@ 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"
@@ -77,6 +81,7 @@ import (
 )
 
 var StorageCounterDSPrefix = "/storage/nextid"
+var dagStore = "dagStore"
 
 func minerAddrFromDS(ds dtypes.MetadataDS) (address.Address, error) {
 	maddrb, err := ds.Get(datastore.NewKey("miner-address"))
@@ -575,6 +580,91 @@ 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,
@@ -586,33 +676,18 @@ func StorageProvider(minerAddress dtypes.MinerAddress,
 	dsw *dagstore.Wrapper,
 ) (storagemarket.StorageProvider, error) {
 	net := smnet.NewFromLibp2pHost(h)
-
-	// 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))
+	store, err := piecefilestore.NewLocalFileStore(piecefilestore.OsPath(r.Path()))
 	if err != nil {
 		return nil, err
 	}
 
-	opt := storageimpl.CustomDealDecisionLogic(storageimpl.DealDeciderFunc(df))
+	dagStorePath := filepath.Join(r.Path(), dagStore)
 
-	return storageimpl.NewProvider(
-		net,
-		namespace.Wrap(ds, datastore.NewKey("/deals/provider")),
-		store,
-		dsw,
-		pieceStore,
-		dataTransfer,
-		spn,
-		address.Address(minerAddress),
-		storedAsk,
-		opt,
-	)
+	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)
 }
 
 func RetrievalDealFilter(userFilter dtypes.RetrievalDealFilter) func(onlineOk dtypes.ConsiderOnlineRetrievalDealsConfigFunc,

node/modules/storageminer_dagstore.go

@@ -1,112 +0,0 @@
-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
-}

scripts/docker-lotus-entrypoint.sh

@@ -5,7 +5,7 @@ if [ ! -z DOCKER_LOTUS_IMPORT_SNAPSHOT ]; then
 	# Don't init if already initialized.
 	if [ ! -f "$GATE" ]; then
 		echo importing minimal snapshot
-		/usr/local/bin/lotus daemon --import-snapshot "$DOCKER_LOTUS_IMPORT_SNAPSHOT" --halt-after-import
+		/usr/local/bin/lotus daemon --import-snapshot "$DOCKER_LOTUS_IMPORT_SNAPSHOT"
 		# Block future inits
 		date > "$GATE"
 	fi