feat: integrate dag store into lotus

node/modules/client.go

@@ -7,7 +7,6 @@ import (
 	"path/filepath"
 	"time"
 
-	"github.com/filecoin-project/lotus/node/repo/retrievalstoremgr"
 	"go.uber.org/fx"
 	"golang.org/x/xerrors"
 
@@ -17,6 +16,7 @@ import (
 	dtgstransport "github.com/filecoin-project/go-data-transfer/transport/graphsync"
 	"github.com/filecoin-project/go-fil-markets/discovery"
 	discoveryimpl "github.com/filecoin-project/go-fil-markets/discovery/impl"
+	"github.com/filecoin-project/go-fil-markets/filestore"
 	"github.com/filecoin-project/go-fil-markets/retrievalmarket"
 	retrievalimpl "github.com/filecoin-project/go-fil-markets/retrievalmarket/impl"
 	rmnet "github.com/filecoin-project/go-fil-markets/retrievalmarket/network"
@@ -40,6 +40,7 @@ import (
 	"github.com/filecoin-project/lotus/node/modules/dtypes"
 	"github.com/filecoin-project/lotus/node/repo"
 	"github.com/filecoin-project/lotus/node/repo/importmgr"
+	"github.com/filecoin-project/lotus/node/repo/retrievalstoremgr"
 )
 
 func HandleMigrateClientFunds(lc fx.Lifecycle, ds dtypes.MetadataDS, wallet full.WalletAPI, fundMgr *market.FundManager) {
@@ -178,22 +179,19 @@ func StorageClient(lc fx.Lifecycle, h host.Host, dataTransfer dtypes.ClientDataT
 	return c, nil
 }
 
-type CARStore struct {
-	dir string
-}
-
-func (c *CARStore) Path(key string) string {
-	return filepath.Join(c.dir, key)
-}
-
 // RetrievalClient creates a new retrieval client attached to the client blockstore
 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) {
 
+	carStore, err := getRetrievalCarStore(r.Path())
+	if err != nil {
+		return nil, err
+	}
+
 	adapter := retrievaladapter.NewRetrievalClientNode(payAPI, chainAPI, stateAPI)
 	network := rmnet.NewFromLibp2pHost(h)
 	client, err := retrievalimpl.NewClient(network,
-		&CARStore{r.Path()}, dt, adapter, resolver, namespace.Wrap(ds, datastore.NewKey("/retrievals/client")))
+		carStore, dt, adapter, resolver, namespace.Wrap(ds, datastore.NewKey("/retrievals/client")))
 	if err != nil {
 		return nil, err
 	}
@@ -211,6 +209,15 @@ func RetrievalClient(lc fx.Lifecycle, h host.Host, r repo.LockedRepo, dt dtypes.
 	return client, nil
 }
 
+func getRetrievalCarStore(path string) (filestore.CarFileStore, error) {
+	carStorePath := filepath.Join(path, "retrieval-cars")
+	err := os.Mkdir(carStorePath, os.ModePerm)
+	if err != nil {
+		return nil, xerrors.Errorf("could not create directory %s: %w", carStorePath, err)
+	}
+	return filestore.NewLocalCarStore(path)
+}
+
 // ClientBlockstoreRetrievalStoreManager is the default version of the RetrievalStoreManager that runs on multistore
 func ClientBlockstoreRetrievalStoreManager(isIpfsRetrieval bool) func(bs dtypes.ClientBlockstore) (dtypes.ClientRetrievalStoreManager, error) {
 	return func(bs dtypes.ClientBlockstore) (dtypes.ClientRetrievalStoreManager, error) {

node/modules/storageminer.go

@@ -11,11 +11,6 @@ import (
 	"strings"
 	"time"
 
-	mktdagstore "github.com/filecoin-project/go-fil-markets/dagstore"
-
-	"github.com/filecoin-project/go-fil-markets/shared_testutil/dagstore"
-
-	"github.com/filecoin-project/lotus/markets/pricing"
 	"go.uber.org/fx"
 	"go.uber.org/multierr"
 	"golang.org/x/xerrors"
@@ -33,10 +28,13 @@ import (
 	"github.com/libp2p/go-libp2p-core/host"
 	"github.com/libp2p/go-libp2p-core/routing"
 
+	"github.com/filecoin-project/dagstore"
+	"github.com/filecoin-project/dagstore/mount"
 	"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"
+	mktdagstore "github.com/filecoin-project/go-fil-markets/dagstore"
 	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"
@@ -53,13 +51,13 @@ import (
 	"github.com/filecoin-project/go-statestore"
 	"github.com/filecoin-project/go-storedcounter"
 
-	"github.com/filecoin-project/lotus/api"
 	sectorstorage "github.com/filecoin-project/lotus/extern/sector-storage"
 	"github.com/filecoin-project/lotus/extern/sector-storage/ffiwrapper"
 	"github.com/filecoin-project/lotus/extern/sector-storage/stores"
 	sealing "github.com/filecoin-project/lotus/extern/storage-sealing"
 	"github.com/filecoin-project/lotus/extern/storage-sealing/sealiface"
 
+	"github.com/filecoin-project/lotus/api"
 	"github.com/filecoin-project/lotus/api/v0api"
 	"github.com/filecoin-project/lotus/api/v1api"
 	"github.com/filecoin-project/lotus/blockstore"
@@ -71,6 +69,7 @@ import (
 	"github.com/filecoin-project/lotus/journal"
 	"github.com/filecoin-project/lotus/markets"
 	marketevents "github.com/filecoin-project/lotus/markets/loggers"
+	"github.com/filecoin-project/lotus/markets/pricing"
 	lotusminer "github.com/filecoin-project/lotus/miner"
 	"github.com/filecoin-project/lotus/node/config"
 	"github.com/filecoin-project/lotus/node/modules/dtypes"
@@ -569,9 +568,29 @@ func BasicDealFilter(user dtypes.StorageDealFilter) func(onlineOk dtypes.Conside
 	}
 }
 
-func DAGStore(r repo.LockedRepo) (dagstore.DagStore, error) {
-	md := dagstore.NewMockDagStore()
-	return md, nil
+func DAGStoreRegistry() *mount.Registry {
+	return mount.NewRegistry()
+}
+
+func DAGStore(r repo.LockedRepo, ds dtypes.MetadataDS, dsRegistry *mount.Registry) (*dagstore.DAGStore, error) {
+	dagStoreDir := filepath.Join(r.Path(), "dagstore")
+	dagStoreDS := namespace.Wrap(ds, datastore.NewKey("/dagstore/provider"))
+	return dagstore.NewDAGStore(dagstore.Config{
+		TransientsDir: filepath.Join(dagStoreDir, "transients"),
+		IndexDir:      filepath.Join(dagStoreDir, "index"),
+		Datastore:     dagStoreDS,
+		MountRegistry: dsRegistry,
+	})
+}
+
+func DAGStoreWrapper(
+	pieceStore dtypes.ProviderPieceStore,
+	rpn retrievalmarket.RetrievalProviderNode,
+	dsRegistry *mount.Registry,
+	dagStore *dagstore.DAGStore,
+) (mktdagstore.DagStoreWrapper, error) {
+	mountApi := mktdagstore.NewLotusMountAPI(pieceStore, rpn)
+	return mktdagstore.NewDagStoreWrapper(dsRegistry, dagStore, mountApi)
 }
 
 func StorageProvider(minerAddress dtypes.MinerAddress,
@@ -582,7 +601,7 @@ func StorageProvider(minerAddress dtypes.MinerAddress,
 	dataTransfer dtypes.ProviderDataTransfer,
 	spn storagemarket.StorageProviderNode,
 	df dtypes.StorageDealFilter,
-	dagStore dagstore.DagStore,
+	dagStore mktdagstore.DagStoreWrapper,
 ) (storagemarket.StorageProvider, error) {
 	net := smnet.NewFromLibp2pHost(h)
 	store, err := piecefilestore.NewLocalFileStore(piecefilestore.OsPath(r.Path()))
@@ -657,7 +676,7 @@ func RetrievalProvider(
 	dt dtypes.ProviderDataTransfer,
 	pricingFnc dtypes.RetrievalPricingFunc,
 	userFilter dtypes.RetrievalDealFilter,
-	dagStore dagstore.DagStore,
+	dagStore mktdagstore.DagStoreWrapper,
 ) (retrievalmarket.RetrievalProvider, error) {
 	opt := retrievalimpl.DealDeciderOpt(retrievalimpl.DealDecider(userFilter))
 	return retrievalimpl.NewProvider(
@@ -669,7 +688,6 @@ func RetrievalProvider(
 		dt,
 		namespace.Wrap(ds, datastore.NewKey("/retrievals/provider")),
 		retrievalimpl.RetrievalPricingFunc(pricingFnc),
-		mktdagstore.NewMount(pieceStore, adapter),
 		opt,
 	)
 }