update go-fil-markets.

Closes https://github.com/filecoin-project/lotus/issues/7017.

api/api_storage.go

@@ -398,6 +398,7 @@ type DagstoreShardResult struct {
 
 type DagstoreInitializeAllParams struct {
 	MaxConcurrency int
+	IncludeSealed  bool
 }
 
 // DagstoreInitializeAllEvent represents an initialization event.

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-m1.3.5"
 
 func UserVersion() string {
 	if os.Getenv("LOTUS_VERSION_IGNORE_COMMIT") == "1" {

cmd/lotus-miner/dagstore.go

@@ -27,7 +27,18 @@ var dagstoreCmd = &cli.Command{
 var dagstoreListShardsCmd = &cli.Command{
 	Name:  "list-shards",
 	Usage: "List all shards known to the dagstore, with their current status",
+	Flags: []cli.Flag{
+		&cli.BoolFlag{
+			Name:        "color",
+			Usage:       "use color in display output",
+			DefaultText: "depends on output being a TTY",
+		},
+	},
 	Action: func(cctx *cli.Context) error {
+		if cctx.IsSet("color") {
+			color.NoColor = !cctx.Bool("color")
+		}
+
 		marketsApi, closer, err := lcli.GetMarketsAPI(cctx)
 		if err != nil {
 			return err
@@ -80,7 +91,18 @@ var dagstoreInitializeShardCmd = &cli.Command{
 	Name:      "initialize-shard",
 	ArgsUsage: "[key]",
 	Usage:     "Initialize the specified shard",
+	Flags: []cli.Flag{
+		&cli.BoolFlag{
+			Name:        "color",
+			Usage:       "use color in display output",
+			DefaultText: "depends on output being a TTY",
+		},
+	},
 	Action: func(cctx *cli.Context) error {
+		if cctx.IsSet("color") {
+			color.NoColor = !cctx.Bool("color")
+		}
+
 		if cctx.NArg() != 1 {
 			return fmt.Errorf("must provide a single shard key")
 		}
@@ -101,7 +123,18 @@ var dagstoreRecoverShardCmd = &cli.Command{
 	Name:      "recover-shard",
 	ArgsUsage: "[key]",
 	Usage:     "Attempt to recover a shard in errored state",
+	Flags: []cli.Flag{
+		&cli.BoolFlag{
+			Name:        "color",
+			Usage:       "use color in display output",
+			DefaultText: "depends on output being a TTY",
+		},
+	},
 	Action: func(cctx *cli.Context) error {
+		if cctx.IsSet("color") {
+			color.NoColor = !cctx.Bool("color")
+		}
+
 		if cctx.NArg() != 1 {
 			return fmt.Errorf("must provide a single shard key")
 		}
@@ -120,16 +153,30 @@ var dagstoreRecoverShardCmd = &cli.Command{
 
 var dagstoreInitializeAllCmd = &cli.Command{
 	Name:  "initialize-all",
-	Usage: "Initialize all uninitialized shards, streaming results as they're produced",
+	Usage: "Initialize all uninitialized shards, streaming results as they're produced; only shards for unsealed pieces are initialized by default",
 	Flags: []cli.Flag{
 		&cli.UintFlag{
 			Name:     "concurrency",
 			Usage:    "maximum shards to initialize concurrently at a time; use 0 for unlimited",
 			Required: true,
 		},
+		&cli.BoolFlag{
+			Name:  "include-sealed",
+			Usage: "initialize sealed pieces as well",
+		},
+		&cli.BoolFlag{
+			Name:        "color",
+			Usage:       "use color in display output",
+			DefaultText: "depends on output being a TTY",
+		},
 	},
 	Action: func(cctx *cli.Context) error {
+		if cctx.IsSet("color") {
+			color.NoColor = !cctx.Bool("color")
+		}
+
 		concurrency := cctx.Uint("concurrency")
+		sealed := cctx.Bool("sealed")
 
 		marketsApi, closer, err := lcli.GetMarketsAPI(cctx)
 		if err != nil {
@@ -141,6 +188,7 @@ var dagstoreInitializeAllCmd = &cli.Command{
 
 		params := api.DagstoreInitializeAllParams{
 			MaxConcurrency: int(concurrency),
+			IncludeSealed:  sealed,
 		}
 
 		ch, err := marketsApi.DagstoreInitializeAll(ctx, params)
@@ -176,8 +224,18 @@ var dagstoreInitializeAllCmd = &cli.Command{
 var dagstoreGcCmd = &cli.Command{
 	Name:  "gc",
 	Usage: "Garbage collect the dagstore",
-
+	Flags: []cli.Flag{
+		&cli.BoolFlag{
+			Name:        "color",
+			Usage:       "use color in display output",
+			DefaultText: "depends on output being a TTY",
+		},
+	},
 	Action: func(cctx *cli.Context) error {
+		if cctx.IsSet("color") {
+			color.NoColor = !cctx.Bool("color")
+		}
+
 		marketsApi, closer, err := lcli.GetMarketsAPI(cctx)
 		if err != nil {
 			return err

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

@@ -380,7 +380,8 @@ Inputs:
 ```json
 [
   {
-    "MaxConcurrency": 123
+    "MaxConcurrency": 123,
+    "IncludeSealed": true
   }
 ]
 ```

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-m1.3.5
 
 COMMANDS:
    init     Initialize a lotus miner repo
@@ -1013,7 +1013,7 @@ 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
+   initialize-all    Initialize all uninitialized shards, streaming results as they're produced; only shards for unsealed pieces are initialized by default
    gc                Garbage collect the dagstore
    help, h           Shows a list of commands or help for one command
 
@@ -1065,13 +1065,14 @@ OPTIONS:
 ### lotus-miner dagstore initialize-all
 ```
 NAME:
-   lotus-miner dagstore initialize-all - Initialize all uninitialized shards, streaming results as they're produced
+   lotus-miner dagstore initialize-all - Initialize all uninitialized shards, streaming results as they're produced; only shards for unsealed pieces are initialized by default
 
 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)
+   --include-sealed     initialize sealed pieces as well (default: false)
    --help, -h           show help (default: false)
    
 ```

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-m1.3.5
 
 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-m1.3.5
 
 COMMANDS:
    daemon   Start a lotus daemon process

go.mod

@@ -36,7 +36,7 @@ require (
 	github.com/filecoin-project/go-data-transfer v1.7.3
 	github.com/filecoin-project/go-fil-commcid v0.1.0
 	github.com/filecoin-project/go-fil-commp-hashhash v0.1.0
-	github.com/filecoin-project/go-fil-markets v1.6.3-0.20210806151421-967d8717c917
+	github.com/filecoin-project/go-fil-markets v1.6.3-0.20210811182620-1237843e237c
 	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
@@ -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.2
 	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

@@ -289,8 +289,8 @@ github.com/filecoin-project/go-fil-commcid v0.1.0/go.mod h1:Eaox7Hvus1JgPrL5+M3+
 github.com/filecoin-project/go-fil-commp-hashhash v0.1.0 h1:imrrpZWEHRnNqqv0tN7LXep5bFEVOVmQWHJvl2mgsGo=
 github.com/filecoin-project/go-fil-commp-hashhash v0.1.0/go.mod h1:73S8WSEWh9vr0fDJVnKADhfIv/d6dCbAGaAGWbdJEI8=
 github.com/filecoin-project/go-fil-markets v1.0.5-0.20201113164554-c5eba40d5335/go.mod h1:AJySOJC00JRWEZzRG2KsfUnqEf5ITXxeX09BE9N4f9c=
-github.com/filecoin-project/go-fil-markets v1.6.3-0.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.20210811182620-1237843e237c h1:rHmpzPGgYbr9efLWkRHahkPyjrVCMesVGB8G8vxFmEg=
+github.com/filecoin-project/go-fil-markets v1.6.3-0.20210811182620-1237843e237c/go.mod h1:D5xHWxyuU0EK8wcK4qStO5rjmpH206eb4OdrkWmTdaY=
 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=
@@ -760,8 +760,8 @@ github.com/ipld/go-car v0.1.1-0.20200923150018-8cdef32e2da4/go.mod h1:xrMEcuSq+D
 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/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-car/v2 v2.0.2 h1:R1oIAPwrGp26mEFzcGf5bfTZAAHDOkaVnZTEVebaWX4=
+github.com/ipld/go-car/v2 v2.0.2/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=

markets/dagstore/miner_api.go

@@ -23,17 +23,17 @@ type MinerAPI interface {
 
 type minerAPI struct {
 	pieceStore piecestore.PieceStore
-	rm         retrievalmarket.RetrievalProviderNode
+	sa         retrievalmarket.SectorAccessor
 	throttle   throttle.Throttler
 	readyMgr   *shared.ReadyManager
 }
 
 var _ MinerAPI = (*minerAPI)(nil)
 
-func NewMinerAPI(store piecestore.PieceStore, rm retrievalmarket.RetrievalProviderNode, concurrency int) MinerAPI {
+func NewMinerAPI(store piecestore.PieceStore, sa retrievalmarket.SectorAccessor, concurrency int) MinerAPI {
 	return &minerAPI{
 		pieceStore: store,
-		rm:         rm,
+		sa:         sa,
 		throttle:   throttle.Fixed(concurrency),
 		readyMgr:   shared.NewReadyManager(),
 	}
@@ -70,7 +70,7 @@ func (m *minerAPI) IsUnsealed(ctx context.Context, pieceCid cid.Cid) (bool, erro
 		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())
+			isUnsealed, err = m.sa.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)
 			}
@@ -119,7 +119,7 @@ func (m *minerAPI) FetchUnsealedPiece(ctx context.Context, pieceCid cid.Cid) (io
 		// Throttle this path to avoid flooding the storage subsystem.
 		var reader io.ReadCloser
 		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())
+			isUnsealed, err := m.sa.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)
 			}
@@ -127,7 +127,7 @@ func (m *minerAPI) FetchUnsealedPiece(ctx context.Context, pieceCid cid.Cid) (io
 				return nil
 			}
 			// Because we know we have an unsealed copy, this UnsealSector call will actually not perform any unsealing.
-			reader, err = m.rm.UnsealSector(ctx, deal.SectorID, deal.Offset.Unpadded(), deal.Length.Unpadded())
+			reader, err = m.sa.UnsealSector(ctx, deal.SectorID, deal.Offset.Unpadded(), deal.Length.Unpadded())
 			return err
 		})
 
@@ -149,7 +149,7 @@ func (m *minerAPI) FetchUnsealedPiece(ctx context.Context, pieceCid cid.Cid) (io
 		// block for a long time with the current PoRep
 		//
 		// This path is unthrottled.
-		reader, err := m.rm.UnsealSector(ctx, deal.SectorID, deal.Offset.Unpadded(), deal.Length.Unpadded())
+		reader, err := m.sa.UnsealSector(ctx, deal.SectorID, deal.Offset.Unpadded(), deal.Length.Unpadded())
 		if err != nil {
 			lastErr = xerrors.Errorf("failed to unseal deal %d: %w", deal.DealID, err)
 			log.Warn(lastErr.Error())

markets/dagstore/wrapper.go

@@ -53,10 +53,10 @@ type Wrapper struct {
 
 var _ stores.DAGStoreWrapper = (*Wrapper)(nil)
 
-func NewDAGStore(cfg config.DAGStoreConfig, mountApi MinerAPI) (*dagstore.DAGStore, *Wrapper, error) {
+func NewDAGStore(cfg config.DAGStoreConfig, minerApi MinerAPI) (*dagstore.DAGStore, *Wrapper, error) {
 	// construct the DAG Store.
 	registry := mount.NewRegistry()
-	if err := registry.Register(lotusScheme, mountTemplate(mountApi)); err != nil {
+	if err := registry.Register(lotusScheme, mountTemplate(minerApi)); err != nil {
 		return nil, nil, xerrors.Errorf("failed to create registry: %w", err)
 	}
 
@@ -104,7 +104,7 @@ func NewDAGStore(cfg config.DAGStoreConfig, mountApi MinerAPI) (*dagstore.DAGSto
 	w := &Wrapper{
 		cfg:        cfg,
 		dagst:      dagst,
-		minerAPI:   mountApi,
+		minerAPI:   minerApi,
 		failureCh:  failureCh,
 		traceCh:    traceCh,
 		gcInterval: time.Duration(cfg.GCInterval),
@@ -331,13 +331,10 @@ func (w *Wrapper) MigrateDeals(ctx context.Context, deals []storagemarket.MinerD
 	// 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
-		}
+		pieceCid := deal.Proposal.PieceCID
 
 		// enrich log statements in this iteration with deal ID and piece CID.
-		log := log.With("deal_id", deal.DealID, "piece_cid", deal.Ref.PieceCid)
+		log := log.With("deal_id", deal.DealID, "piece_cid", pieceCid)
 
 		// Filter for deals that have been handed off to the sealing subsystem
 		if _, ok := inSealingSubsystem[deal.State]; !ok {
@@ -350,7 +347,7 @@ func (w *Wrapper) MigrateDeals(ctx context.Context, deals []storagemarket.MinerD
 		// 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)
+		err = w.RegisterShard(ctx, pieceCid, "", false, resch)
 		if err != nil {
 			log.Warnw("failed to register shard", "error", err)
 			continue

markets/dagstore/wrapper_migration_test.go

@@ -12,12 +12,15 @@ import (
 	"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"
+
+	"github.com/filecoin-project/specs-actors/v2/actors/builtin/market"
 )
 
 func TestShardRegistration(t *testing.T) {
 	ps := tut.NewTestPieceStore()
-	providerNode := testnodes.NewTestRetrievalProviderNode()
+	sa := testnodes.NewTestSectorAccessor()
 
 	ctx := context.Background()
 	cids := tut.GenerateCids(4)
@@ -53,36 +56,44 @@ func TestShardRegistration(t *testing.T) {
 		// Should be registered
 		State:        storagemarket.StorageDealSealing,
 		SectorNumber: unsealedSector1,
-		Ref: &storagemarket.DataRef{
-			PieceCid: &pieceCidUnsealed,
+		ClientDealProposal: market.ClientDealProposal{
+			Proposal: market.DealProposal{
+				PieceCID: pieceCidUnsealed,
+			},
 		},
 	}, {
 		// Should be registered with lazy registration (because sector is sealed)
 		State:        storagemarket.StorageDealSealing,
 		SectorNumber: sealedSector,
-		Ref: &storagemarket.DataRef{
-			PieceCid: &pieceCidSealed,
+		ClientDealProposal: market.ClientDealProposal{
+			Proposal: market.DealProposal{
+				PieceCID: pieceCidSealed,
+			},
 		},
 	}, {
 		// Should be ignored because deal is no longer active
 		State:        storagemarket.StorageDealError,
 		SectorNumber: unsealedSector2,
-		Ref: &storagemarket.DataRef{
-			PieceCid: &pieceCidUnsealed2,
+		ClientDealProposal: market.ClientDealProposal{
+			Proposal: market.DealProposal{
+				PieceCID: pieceCidUnsealed2,
+			},
 		},
 	}, {
 		// Should be ignored because deal is not yet sealing
 		State:        storagemarket.StorageDealFundsReserved,
 		SectorNumber: unsealedSector3,
-		Ref: &storagemarket.DataRef{
-			PieceCid: &pieceCidUnsealed3,
+		ClientDealProposal: market.ClientDealProposal{
+			Proposal: market.DealProposal{
+				PieceCID: pieceCidUnsealed3,
+			},
 		},
 	}}
 
 	cfg := config.DefaultStorageMiner().DAGStore
 	cfg.RootDir = t.TempDir()
 
-	mapi := NewMinerAPI(ps, providerNode, 10)
+	mapi := NewMinerAPI(ps, sa, 10)
 	dagst, w, err := NewDAGStore(cfg, mapi)
 	require.NoError(t, err)
 	require.NotNil(t, dagst)

markets/retrievaladapter/provider.go

@@ -2,44 +2,34 @@ package retrievaladapter
 
 import (
 	"context"
-	"io"
 
-	"github.com/filecoin-project/lotus/api"
-	"github.com/filecoin-project/lotus/api/v1api"
-	"github.com/filecoin-project/lotus/node/modules/dtypes"
-	"github.com/filecoin-project/lotus/storage/sectorblocks"
 	"github.com/hashicorp/go-multierror"
 	"golang.org/x/xerrors"
 
 	"github.com/ipfs/go-cid"
 
-	"github.com/filecoin-project/lotus/chain/actors/builtin/paych"
-	"github.com/filecoin-project/lotus/chain/types"
-	sectorstorage "github.com/filecoin-project/lotus/extern/sector-storage"
-	"github.com/filecoin-project/lotus/extern/sector-storage/storiface"
-
 	"github.com/filecoin-project/go-address"
 	"github.com/filecoin-project/go-fil-markets/retrievalmarket"
 	"github.com/filecoin-project/go-fil-markets/shared"
 	"github.com/filecoin-project/go-state-types/abi"
-	specstorage "github.com/filecoin-project/specs-storage/storage"
-
+	"github.com/filecoin-project/lotus/api/v1api"
+	"github.com/filecoin-project/lotus/chain/actors/builtin/paych"
+	"github.com/filecoin-project/lotus/chain/types"
 	logging "github.com/ipfs/go-log/v2"
 )
 
 var log = logging.Logger("retrievaladapter")
 
 type retrievalProviderNode struct {
-	maddr address.Address
-	secb  sectorblocks.SectorBuilder
-	pp    sectorstorage.PieceProvider
-	full  v1api.FullNode
+	full v1api.FullNode
 }
 
+var _ retrievalmarket.RetrievalProviderNode = (*retrievalProviderNode)(nil)
+
 // NewRetrievalProviderNode returns a new node adapter for a retrieval provider that talks to the
 // Lotus Node
-func NewRetrievalProviderNode(maddr dtypes.MinerAddress, secb sectorblocks.SectorBuilder, pp sectorstorage.PieceProvider, full v1api.FullNode) retrievalmarket.RetrievalProviderNode {
-	return &retrievalProviderNode{address.Address(maddr), secb, pp, full}
+func NewRetrievalProviderNode(full v1api.FullNode) retrievalmarket.RetrievalProviderNode {
+	return &retrievalProviderNode{full: full}
 }
 
 func (rpn *retrievalProviderNode) GetMinerWorkerAddress(ctx context.Context, miner address.Address, tok shared.TipSetToken) (address.Address, error) {
@@ -52,42 +42,6 @@ func (rpn *retrievalProviderNode) GetMinerWorkerAddress(ctx context.Context, min
 	return mi.Worker, err
 }
 
-func (rpn *retrievalProviderNode) UnsealSector(ctx context.Context, sectorID abi.SectorNumber, offset abi.UnpaddedPieceSize, length abi.UnpaddedPieceSize) (io.ReadCloser, error) {
-	log.Debugf("get sector %d, offset %d, length %d", sectorID, offset, length)
-	si, err := rpn.sectorsStatus(ctx, sectorID, false)
-	if err != nil {
-		return nil, err
-	}
-
-	mid, err := address.IDFromAddress(rpn.maddr)
-	if err != nil {
-		return nil, err
-	}
-
-	ref := specstorage.SectorRef{
-		ID: abi.SectorID{
-			Miner:  abi.ActorID(mid),
-			Number: sectorID,
-		},
-		ProofType: si.SealProof,
-	}
-
-	var commD cid.Cid
-	if si.CommD != nil {
-		commD = *si.CommD
-	}
-
-	// Get a reader for the piece, unsealing the piece if necessary
-	log.Debugf("read piece in sector %d, offset %d, length %d from miner %d", sectorID, offset, length, mid)
-	r, unsealed, err := rpn.pp.ReadPiece(ctx, ref, storiface.UnpaddedByteIndex(offset), length, si.Ticket.Value, commD)
-	if err != nil {
-		return nil, xerrors.Errorf("failed to unseal piece from sector %d: %w", sectorID, err)
-	}
-	_ = unsealed // todo: use
-
-	return r, nil
-}
-
 func (rpn *retrievalProviderNode) SavePaymentVoucher(ctx context.Context, paymentChannel address.Address, voucher *paych.SignedVoucher, proof []byte, expectedAmount abi.TokenAmount, tok shared.TipSetToken) (abi.TokenAmount, error) {
 	// TODO: respect the provided TipSetToken (a serialized TipSetKey) when
 	// querying the chain
@@ -104,29 +58,6 @@ func (rpn *retrievalProviderNode) GetChainHead(ctx context.Context) (shared.TipS
 	return head.Key().Bytes(), head.Height(), nil
 }
 
-func (rpn *retrievalProviderNode) IsUnsealed(ctx context.Context, sectorID abi.SectorNumber, offset abi.UnpaddedPieceSize, length abi.UnpaddedPieceSize) (bool, error) {
-	si, err := rpn.sectorsStatus(ctx, sectorID, true)
-	if err != nil {
-		return false, xerrors.Errorf("failed to get sector info: %w", err)
-	}
-
-	mid, err := address.IDFromAddress(rpn.maddr)
-	if err != nil {
-		return false, err
-	}
-
-	ref := specstorage.SectorRef{
-		ID: abi.SectorID{
-			Miner:  abi.ActorID(mid),
-			Number: sectorID,
-		},
-		ProofType: si.SealProof,
-	}
-
-	log.Debugf("will call IsUnsealed now sector=%+v, offset=%d, size=%d", sectorID, offset, length)
-	return rpn.pp.IsUnsealed(ctx, ref, storiface.UnpaddedByteIndex(offset), length)
-}
-
 // GetRetrievalPricingInput takes a set of candidate storage deals that can serve a retrieval request,
 // and returns an minimally populated PricingInput. This PricingInput should be enhanced
 // with more data, and passed to the pricing function to determine the final quoted price.
@@ -175,37 +106,3 @@ func (rpn *retrievalProviderNode) GetRetrievalPricingInput(ctx context.Context,
 
 	return resp, nil
 }
-
-func (rpn *retrievalProviderNode) sectorsStatus(ctx context.Context, sid abi.SectorNumber, showOnChainInfo bool) (api.SectorInfo, error) {
-	sInfo, err := rpn.secb.SectorsStatus(ctx, sid, false)
-	if err != nil {
-		return api.SectorInfo{}, err
-	}
-
-	if !showOnChainInfo {
-		return sInfo, nil
-	}
-
-	onChainInfo, err := rpn.full.StateSectorGetInfo(ctx, rpn.maddr, sid, types.EmptyTSK)
-	if err != nil {
-		return sInfo, err
-	}
-	if onChainInfo == nil {
-		return sInfo, nil
-	}
-	sInfo.SealProof = onChainInfo.SealProof
-	sInfo.Activation = onChainInfo.Activation
-	sInfo.Expiration = onChainInfo.Expiration
-	sInfo.DealWeight = onChainInfo.DealWeight
-	sInfo.VerifiedDealWeight = onChainInfo.VerifiedDealWeight
-	sInfo.InitialPledge = onChainInfo.InitialPledge
-
-	ex, err := rpn.full.StateSectorExpiration(ctx, rpn.maddr, sid, types.EmptyTSK)
-	if err != nil {
-		return sInfo, nil
-	}
-	sInfo.OnTime = ex.OnTime
-	sInfo.Early = ex.Early
-
-	return sInfo, nil
-}

markets/sectoraccessor/sectoraccessor.go

@@ -0,0 +1,132 @@
+package sectoraccessor
+
+import (
+	"context"
+	"io"
+
+	"golang.org/x/xerrors"
+
+	"github.com/filecoin-project/lotus/api"
+	"github.com/filecoin-project/lotus/api/v1api"
+	"github.com/filecoin-project/lotus/chain/types"
+	sectorstorage "github.com/filecoin-project/lotus/extern/sector-storage"
+	"github.com/filecoin-project/lotus/extern/sector-storage/storiface"
+	"github.com/filecoin-project/lotus/node/modules/dtypes"
+	"github.com/filecoin-project/lotus/storage/sectorblocks"
+
+	"github.com/filecoin-project/go-address"
+	"github.com/filecoin-project/go-fil-markets/retrievalmarket"
+	"github.com/filecoin-project/go-state-types/abi"
+	specstorage "github.com/filecoin-project/specs-storage/storage"
+
+	"github.com/ipfs/go-cid"
+	logging "github.com/ipfs/go-log/v2"
+)
+
+var log = logging.Logger("sectoraccessor")
+
+type sectorAccessor struct {
+	maddr address.Address
+	secb  sectorblocks.SectorBuilder
+	pp    sectorstorage.PieceProvider
+	full  v1api.FullNode
+}
+
+var _ retrievalmarket.SectorAccessor = (*sectorAccessor)(nil)
+
+func NewSectorAccessor(maddr dtypes.MinerAddress, secb sectorblocks.SectorBuilder, pp sectorstorage.PieceProvider, full v1api.FullNode) retrievalmarket.SectorAccessor {
+	return &sectorAccessor{address.Address(maddr), secb, pp, full}
+}
+
+func (sa *sectorAccessor) UnsealSector(ctx context.Context, sectorID abi.SectorNumber, offset abi.UnpaddedPieceSize, length abi.UnpaddedPieceSize) (io.ReadCloser, error) {
+	log.Debugf("get sector %d, offset %d, length %d", sectorID, offset, length)
+	si, err := sa.sectorsStatus(ctx, sectorID, false)
+	if err != nil {
+		return nil, err
+	}
+
+	mid, err := address.IDFromAddress(sa.maddr)
+	if err != nil {
+		return nil, err
+	}
+
+	ref := specstorage.SectorRef{
+		ID: abi.SectorID{
+			Miner:  abi.ActorID(mid),
+			Number: sectorID,
+		},
+		ProofType: si.SealProof,
+	}
+
+	var commD cid.Cid
+	if si.CommD != nil {
+		commD = *si.CommD
+	}
+
+	// Get a reader for the piece, unsealing the piece if necessary
+	log.Debugf("read piece in sector %d, offset %d, length %d from miner %d", sectorID, offset, length, mid)
+	r, unsealed, err := sa.pp.ReadPiece(ctx, ref, storiface.UnpaddedByteIndex(offset), length, si.Ticket.Value, commD)
+	if err != nil {
+		return nil, xerrors.Errorf("failed to unseal piece from sector %d: %w", sectorID, err)
+	}
+	_ = unsealed // todo: use
+
+	return r, nil
+}
+
+func (sa *sectorAccessor) IsUnsealed(ctx context.Context, sectorID abi.SectorNumber, offset abi.UnpaddedPieceSize, length abi.UnpaddedPieceSize) (bool, error) {
+	si, err := sa.sectorsStatus(ctx, sectorID, true)
+	if err != nil {
+		return false, xerrors.Errorf("failed to get sector info: %w", err)
+	}
+
+	mid, err := address.IDFromAddress(sa.maddr)
+	if err != nil {
+		return false, err
+	}
+
+	ref := specstorage.SectorRef{
+		ID: abi.SectorID{
+			Miner:  abi.ActorID(mid),
+			Number: sectorID,
+		},
+		ProofType: si.SealProof,
+	}
+
+	log.Debugf("will call IsUnsealed now sector=%+v, offset=%d, size=%d", sectorID, offset, length)
+	return sa.pp.IsUnsealed(ctx, ref, storiface.UnpaddedByteIndex(offset), length)
+}
+
+func (sa *sectorAccessor) sectorsStatus(ctx context.Context, sid abi.SectorNumber, showOnChainInfo bool) (api.SectorInfo, error) {
+	sInfo, err := sa.secb.SectorsStatus(ctx, sid, false)
+	if err != nil {
+		return api.SectorInfo{}, err
+	}
+
+	if !showOnChainInfo {
+		return sInfo, nil
+	}
+
+	onChainInfo, err := sa.full.StateSectorGetInfo(ctx, sa.maddr, sid, types.EmptyTSK)
+	if err != nil {
+		return sInfo, err
+	}
+	if onChainInfo == nil {
+		return sInfo, nil
+	}
+	sInfo.SealProof = onChainInfo.SealProof
+	sInfo.Activation = onChainInfo.Activation
+	sInfo.Expiration = onChainInfo.Expiration
+	sInfo.DealWeight = onChainInfo.DealWeight
+	sInfo.VerifiedDealWeight = onChainInfo.VerifiedDealWeight
+	sInfo.InitialPledge = onChainInfo.InitialPledge
+
+	ex, err := sa.full.StateSectorExpiration(ctx, sa.maddr, sid, types.EmptyTSK)
+	if err != nil {
+		return sInfo, nil
+	}
+	sInfo.OnTime = ex.OnTime
+	sInfo.Early = ex.Early
+
+	return sInfo, nil
+}

markets/storageadapter/provider.go

@@ -7,7 +7,6 @@ import (
 	"io"
 	"time"
 
-	"github.com/filecoin-project/go-fil-markets/retrievalmarket"
 	"github.com/ipfs/go-cid"
 	logging "github.com/ipfs/go-log/v2"
 	"go.uber.org/fx"
@@ -58,12 +57,10 @@ type ProviderNodeAdapter struct {
 	maxDealCollateralMultiplier uint64
 	dsMatcher                   *dealStateMatcher
 	scMgr                       *SectorCommittedManager
-
-	rpn retrievalmarket.RetrievalProviderNode
 }
 
-func NewProviderNodeAdapter(fc *config.MinerFeeConfig, dc *config.DealmakingConfig) func(mctx helpers.MetricsCtx, lc fx.Lifecycle, dag dtypes.StagingDAG, secb *sectorblocks.SectorBlocks, full v1api.FullNode, dealPublisher *DealPublisher, rpn retrievalmarket.RetrievalProviderNode) storagemarket.StorageProviderNode {
-	return func(mctx helpers.MetricsCtx, lc fx.Lifecycle, dag dtypes.StagingDAG, secb *sectorblocks.SectorBlocks, full v1api.FullNode, dealPublisher *DealPublisher, rpn retrievalmarket.RetrievalProviderNode) storagemarket.StorageProviderNode {
+func NewProviderNodeAdapter(fc *config.MinerFeeConfig, dc *config.DealmakingConfig) func(mctx helpers.MetricsCtx, lc fx.Lifecycle, dag dtypes.StagingDAG, secb *sectorblocks.SectorBlocks, full v1api.FullNode, dealPublisher *DealPublisher) storagemarket.StorageProviderNode {
+	return func(mctx helpers.MetricsCtx, lc fx.Lifecycle, dag dtypes.StagingDAG, secb *sectorblocks.SectorBlocks, full v1api.FullNode, dealPublisher *DealPublisher) storagemarket.StorageProviderNode {
 		ctx := helpers.LifecycleCtx(mctx, lc)
 
 		ev := events.NewEvents(ctx, full)
@@ -84,7 +81,6 @@ func NewProviderNodeAdapter(fc *config.MinerFeeConfig, dc *config.DealmakingConf
 			na.maxDealCollateralMultiplier = dc.MaxProviderCollateralMultiplier
 		}
 		na.scMgr = NewSectorCommittedManager(ev, na, &apiWrapper{api: full})
-		na.rpn = rpn
 
 		return na
 	}
@@ -426,8 +422,4 @@ func (n *ProviderNodeAdapter) OnDealExpiredOrSlashed(ctx context.Context, dealID
 	return nil
 }
 
-func (n *ProviderNodeAdapter) IsUnsealed(ctx context.Context, sectorID abi.SectorNumber, offset abi.UnpaddedPieceSize, length abi.UnpaddedPieceSize) (bool, error) {
-	return n.rpn.IsUnsealed(ctx, sectorID, offset, length)
-}
-
 var _ storagemarket.StorageProviderNode = &ProviderNodeAdapter{}

node/builder_miner.go

@@ -25,6 +25,7 @@ import (
 	"github.com/filecoin-project/lotus/markets/dagstore"
 	"github.com/filecoin-project/lotus/markets/dealfilter"
 	"github.com/filecoin-project/lotus/markets/retrievaladapter"
+	"github.com/filecoin-project/lotus/markets/sectoraccessor"
 	"github.com/filecoin-project/lotus/markets/storageadapter"
 	"github.com/filecoin-project/lotus/miner"
 	"github.com/filecoin-project/lotus/node/config"
@@ -152,6 +153,7 @@ func ConfigStorageMiner(c interface{}) Option {
 			Override(DAGStoreKey, modules.DAGStore),
 
 			// Markets (retrieval)
+			Override(new(retrievalmarket.SectorAccessor), sectoraccessor.NewSectorAccessor),
 			Override(new(retrievalmarket.RetrievalProviderNode), retrievaladapter.NewRetrievalProviderNode),
 			Override(new(rmnet.RetrievalMarketNetwork), modules.RetrievalNetwork),
 			Override(new(retrievalmarket.RetrievalProvider), modules.RetrievalProvider),

node/config/doc_gen.go

@@ -130,14 +130,23 @@ and storage providers`,
 			Name: "RootDir",
 			Type: "string",
 
-			Comment: `Default value: $LOTUS_MARKETS_PATH/dagStore (split deployment) or
-$LOTUS_MINER_PATH/dagStore (monolith deployment)`,
+			Comment: `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)`,
 		},
 		{
 			Name: "MaxConcurrentIndex",
 			Type: "int",
 
 			Comment: `The maximum amount of indexing jobs that can run simultaneously.
+0 means unlimited.
 Default value: 5.`,
 		},
 		{
@@ -145,8 +154,8 @@ Default value: 5.`,
 			Type: "int",
 
 			Comment: `The maximum amount of unsealed deals that can be fetched simultaneously
-from the storage subsystem.
-Default value: 2.`,
+from the storage subsystem. 0 means unlimited.
+Default value: 0 (unlimited).`,
 		},
 		{
 			Name: "MaxConcurrencyStorageCalls",

node/config/types.go

@@ -57,24 +57,23 @@ 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)
+	// 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.
+	// 0 means unlimited.
 	// Default value: 5.
 	MaxConcurrentIndex int
 
 	// The maximum amount of unsealed deals that can be fetched simultaneously
-	// from the storage subsystem.
-	// Default value: 2.
+	// from the storage subsystem. 0 means unlimited.
+	// Default value: 0 (unlimited).
 	MaxConcurrentReadyFetches int
 
 	// The maximum number of simultaneous inflight API calls to the storage

node/impl/client/client.go

@@ -7,6 +7,7 @@ import (
 	"io"
 	"math/rand"
 	"os"
+	"path/filepath"
 	"sort"
 	"time"
 
@@ -61,6 +62,7 @@ import (
 	"github.com/filecoin-project/lotus/node/impl/full"
 	"github.com/filecoin-project/lotus/node/impl/paych"
 	"github.com/filecoin-project/lotus/node/modules/dtypes"
+	"github.com/filecoin-project/lotus/node/repo"
 	"github.com/filecoin-project/lotus/node/repo/importmgr"
 )
 
@@ -68,6 +70,7 @@ var DefaultHashFunction = uint64(mh.BLAKE2B_MIN + 31)
 
 // 8 days ~=  SealDuration + PreCommit + MaxProveCommitDuration + 8 hour buffer
 const dealStartBufferHours uint64 = 8 * 24
+const DefaultDAGStoreDir = "dagstore"
 
 type API struct {
 	fx.In
@@ -88,6 +91,7 @@ type API struct {
 	Host         host.Host
 
 	RetrievalStoreMgr dtypes.ClientRetrievalStoreManager
+	Repo              repo.LockedRepo
 }
 
 func calcDealExpiration(minDuration uint64, md *dline.Info, startEpoch abi.ChainEpoch) abi.ChainEpoch {
@@ -779,6 +783,13 @@ func (a *API) clientRetrieve(ctx context.Context, order api.RetrievalOrder, ref
 			}
 		})
 
+		tmpCarv2FilePath, err := a.getTmpCarV2FilePath()
+		if err != nil {
+			unsubscribe()
+			finish(xerrors.Errorf("Retrieve failed: %w", err))
+			return
+		}
+
 		resp, err := a.Retrieval.Retrieve(
 			ctx,
 			order.Root,
@@ -786,7 +797,8 @@ func (a *API) clientRetrieve(ctx context.Context, order api.RetrievalOrder, ref
 			order.Total,
 			*order.MinerPeer,
 			order.Client,
-			order.Miner)
+			order.Miner,
+			tmpCarv2FilePath)
 
 		if err != nil {
 			unsubscribe()
@@ -887,6 +899,17 @@ func (a *API) clientRetrieve(ctx context.Context, order api.RetrievalOrder, ref
 	return
 }
 
+// TODO: Come up with a better mechanism for creating the tmp CARv2 file path
+func (a *API) getTmpCarV2FilePath() (string, error) {
+	carsPath := filepath.Join(a.Repo.Path(), DefaultDAGStoreDir, "retrieval-cars")
+
+	if err := os.MkdirAll(carsPath, 0755); err != nil {
+		return "", xerrors.Errorf("failed to create dir")
+	}
+
+	return filepath.Join(carsPath, fmt.Sprintf("%d.car", time.Now().UnixNano())), nil
+}
+
 func (a *API) ClientListRetrievals(ctx context.Context) ([]api.RetrievalInfo, error) {
 	deals, err := a.Retrieval.ListDeals()
 	if err != nil {

node/impl/client/import.go

@@ -16,8 +16,6 @@ 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"
@@ -63,11 +61,9 @@ func (a *API) doImport(ctx context.Context, src string, dst string) (cid.Cid, er
 
 	// 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))
+	bs, err := stores.ReadWriteFilestore(dst, 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 filestore: %w", err)
 	}
 
 	bsvc = blockservice.New(bs, offline.Exchange(bs))
@@ -78,7 +74,7 @@ func (a *API) doImport(ctx context.Context, src string, dst string) (cid.Cid, er
 		return cid.Undef, xerrors.Errorf("failed to create UnixFS DAG with carv2 blockstore: %w", err)
 	}
 
-	if err := bs.Finalize(); err != nil {
+	if err := bs.Close(); err != nil {
 		return cid.Undef, xerrors.Errorf("failed to finalize car blockstore: %w", err)
 	}
 

node/impl/storminer.go

@@ -68,6 +68,7 @@ type StorageMinerAPI struct {
 	PieceStore        dtypes.ProviderPieceStore         `optional:"true"`
 	StorageProvider   storagemarket.StorageProvider     `optional:"true"`
 	RetrievalProvider retrievalmarket.RetrievalProvider `optional:"true"`
+	SectorAccessor    retrievalmarket.SectorAccessor    `optional:"true"`
 	DataTransfer      dtypes.ProviderDataTransfer       `optional:"true"`
 	DealPublisher     *storageadapter.DealPublisher     `optional:"true"`
 	SectorBlocks      *sectorblocks.SectorBlocks        `optional:"true"`
@@ -629,6 +630,10 @@ func (sm *StorageMinerAPI) DagstoreInitializeAll(ctx context.Context, params api
 		return nil, fmt.Errorf("dagstore not available on this node")
 	}
 
+	if sm.SectorAccessor == nil {
+		return nil, fmt.Errorf("sector accessor not available on this node")
+	}
+
 	// prepare the thottler tokens.
 	var throttle chan struct{}
 	if c := params.MaxConcurrency; c > 0 {
@@ -638,16 +643,54 @@ func (sm *StorageMinerAPI) DagstoreInitializeAll(ctx context.Context, params api
 		}
 	}
 
+	// are we initializing only unsealed pieces?
+	onlyUnsealed := !params.IncludeSealed
+
 	info := sm.DAGStore.AllShardsInfo()
-	var uninit []string
+	var toInitialize []string
 	for k, i := range info {
 		if i.ShardState != dagstore.ShardStateNew {
 			continue
 		}
-		uninit = append(uninit, k.String())
+
+		// if we're initializing only unsealed pieces, check if there's an
+		// unsealed deal for this piece available.
+		if onlyUnsealed {
+			pieceCid, err := cid.Decode(k.String())
+			if err != nil {
+				log.Warnw("DagstoreInitializeAll: failed to decode shard key as piece CID; skipping", "shard_key", k.String(), "error", err)
+				continue
+			}
+
+			pi, err := sm.PieceStore.GetPieceInfo(pieceCid)
+			if err != nil {
+				log.Warnw("DagstoreInitializeAll: failed to get piece info; skipping", "piece_cid", pieceCid, "error", err)
+				continue
+			}
+
+			var isUnsealed bool
+			for _, d := range pi.Deals {
+				isUnsealed, err = sm.SectorAccessor.IsUnsealed(ctx, d.SectorID, d.Offset.Unpadded(), d.Length.Unpadded())
+				if err != nil {
+					log.Warnw("DagstoreInitializeAll: failed to get unsealed status; skipping deal", "deal_id", d.DealID, "error", err)
+					continue
+				}
+				if isUnsealed {
+					break
+				}
+			}
+
+			if !isUnsealed {
+				log.Infow("DagstoreInitializeAll: skipping piece because it's sealed", "piece_cid", pieceCid, "error", err)
+				continue
+			}
+		}
+
+		// yes, we're initializing this shard.
+		toInitialize = append(toInitialize, k.String())
 	}
 
-	total := len(uninit)
+	total := len(toInitialize)
 	if total == 0 {
 		out := make(chan api.DagstoreInitializeAllEvent)
 		close(out)
@@ -675,7 +718,7 @@ func (sm *StorageMinerAPI) DagstoreInitializeAll(ctx context.Context, params api
 	}()
 
 	go func() {
-		for i, k := range uninit {
+		for i, k := range toInitialize {
 			select {
 			case <-throttle:
 				// acquired a throttle token, proceed.

node/modules/client.go

@@ -182,16 +182,10 @@ 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")
-
-	if err := os.MkdirAll(carsPath, 0755); err != nil {
-		return nil, xerrors.Errorf("failed to create dir")
-	}
-
 	adapter := retrievaladapter.NewRetrievalClientNode(payAPI, chainAPI, stateAPI)
 	network := rmnet.NewFromLibp2pHost(h)
 	client, err := retrievalimpl.NewClient(network,
-		carsPath, dt, adapter, resolver, namespace.Wrap(ds, datastore.NewKey("/retrievals/client")))
+		dt, adapter, resolver, namespace.Wrap(ds, datastore.NewKey("/retrievals/client")))
 	if err != nil {
 		return nil, err
 	}

node/modules/storageminer.go

@@ -670,6 +670,7 @@ func RetrievalPricingFunc(cfg config.DealmakingConfig) func(_ dtypes.ConsiderOnl
 func RetrievalProvider(
 	maddr dtypes.MinerAddress,
 	adapter retrievalmarket.RetrievalProviderNode,
+	sa retrievalmarket.SectorAccessor,
 	netwk rmnet.RetrievalMarketNetwork,
 	ds dtypes.MetadataDS,
 	pieceStore dtypes.ProviderPieceStore,
@@ -682,6 +683,7 @@ func RetrievalProvider(
 	return retrievalimpl.NewProvider(
 		address.Address(maddr),
 		adapter,
+		sa,
 		netwk,
 		pieceStore,
 		dagStore,

node/modules/storageminer_dagstore.go

@@ -7,10 +7,10 @@ import (
 	"path/filepath"
 	"strconv"
 
-	"github.com/filecoin-project/dagstore"
 	"go.uber.org/fx"
 	"golang.org/x/xerrors"
 
+	"github.com/filecoin-project/dagstore"
 	"github.com/filecoin-project/go-fil-markets/retrievalmarket"
 
 	mdagstore "github.com/filecoin-project/lotus/markets/dagstore"
@@ -25,7 +25,7 @@ const (
 )
 
 // 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) {
+func NewMinerAPI(lc fx.Lifecycle, r repo.LockedRepo, pieceStore dtypes.ProviderPieceStore, sa retrievalmarket.SectorAccessor) (mdagstore.MinerAPI, error) {
 	cfg, err := extractDAGStoreConfig(r)
 	if err != nil {
 		return nil, err
@@ -40,7 +40,7 @@ func NewMinerAPI(lc fx.Lifecycle, r repo.LockedRepo, pieceStore dtypes.ProviderP
 		}
 	}
 
-	mountApi := mdagstore.NewMinerAPI(pieceStore, rpn, cfg.MaxConcurrencyStorageCalls)
+	mountApi := mdagstore.NewMinerAPI(pieceStore, sa, cfg.MaxConcurrencyStorageCalls)
 	ready := make(chan error, 1)
 	pieceStore.OnReady(func(err error) {
 		ready <- err