diff --git a/cmd/lotus-storage-miner/init.go b/cmd/lotus-storage-miner/init.go index 127b90856..f834a09a5 100644 --- a/cmd/lotus-storage-miner/init.go +++ b/cmd/lotus-storage-miner/init.go @@ -362,7 +362,7 @@ func migratePreSealMeta(ctx context.Context, api lapi.FullNode, metadata string, buf := make([]byte, binary.MaxVarintLen64) size := binary.PutUvarint(buf, uint64(maxSectorID+1)) - return mds.Put(datastore.NewKey("/storage/nextid"), buf[:size]) + return mds.Put(datastore.NewKey(modules.StorageCounterDSPrefix), buf[:size]) } func findMarketDealID(ctx context.Context, api lapi.FullNode, deal market.DealProposal) (abi.DealID, error) { diff --git a/go.mod b/go.mod index 46d2d3e88..94aa258c7 100644 --- a/go.mod +++ b/go.mod @@ -21,7 +21,7 @@ require ( github.com/filecoin-project/go-crypto v0.0.0-20191218222705-effae4ea9f03 github.com/filecoin-project/go-data-transfer v0.3.0 github.com/filecoin-project/go-fil-commcid v0.0.0-20200208005934-2b8bd03caca5 - github.com/filecoin-project/go-fil-markets v0.2.3 + github.com/filecoin-project/go-fil-markets v0.2.7 github.com/filecoin-project/go-jsonrpc v0.1.1-0.20200520183639-7c6ee2e066b4 github.com/filecoin-project/go-padreader v0.0.0-20200210211231-548257017ca6 github.com/filecoin-project/go-paramfetch v0.0.2-0.20200505180321-973f8949ea8e diff --git a/go.sum b/go.sum index 4a0082045..f8ab93849 100644 --- a/go.sum +++ b/go.sum @@ -164,8 +164,8 @@ github.com/filecoin-project/go-data-transfer v0.3.0/go.mod h1:cONglGP4s/d+IUQw5m github.com/filecoin-project/go-fil-commcid v0.0.0-20200208005934-2b8bd03caca5 h1:yvQJCW9mmi9zy+51xA01Ea2X7/dL7r8eKDPuGUjRmbo= github.com/filecoin-project/go-fil-commcid v0.0.0-20200208005934-2b8bd03caca5/go.mod h1:JbkIgFF/Z9BDlvrJO1FuKkaWsH673/UdFaiVS6uIHlA= github.com/filecoin-project/go-fil-markets v0.0.0-20200114015428-74d100f305f8/go.mod h1:c8NTjvFVy1Ud02mmGDjOiMeawY2t6ALfrrdvAB01FQc= -github.com/filecoin-project/go-fil-markets v0.2.3 h1:00exBcwysQVEx7wvzcdVz9ZT3HLMXKmbQNIz9ktyeO8= -github.com/filecoin-project/go-fil-markets v0.2.3/go.mod h1:LI3VFHse33aU0djAmFQ8+Hg39i0J8ibAoppGu6TbgkA= +github.com/filecoin-project/go-fil-markets v0.2.7 h1:bgdK/e+xW15aVZLtdFLzAHdrx1hqtGF9veg2lstLK6o= +github.com/filecoin-project/go-fil-markets v0.2.7/go.mod h1:LI3VFHse33aU0djAmFQ8+Hg39i0J8ibAoppGu6TbgkA= github.com/filecoin-project/go-jsonrpc v0.1.1-0.20200520183639-7c6ee2e066b4 h1:H8AVYu0MV9m3CSnKMxeILMfh8xJtnqVdXfBF/qbzgu0= github.com/filecoin-project/go-jsonrpc v0.1.1-0.20200520183639-7c6ee2e066b4/go.mod h1:j6zV//WXIIY5kky873Q3iIKt/ViOE8rcijovmpxrXzM= github.com/filecoin-project/go-padreader v0.0.0-20200210211231-548257017ca6 h1:92PET+sx1Hb4W/8CgFwGuxaKbttwY+UNspYZTvXY0vs= @@ -1086,6 +1086,7 @@ github.com/whyrusleeping/go-ctrlnet v0.0.0-20180313164037-f564fbbdaa95/go.mod h1 github.com/whyrusleeping/go-keyspace v0.0.0-20160322163242-5b898ac5add1 h1:EKhdznlJHPMoKr0XTrX+IlJs1LH3lyx2nfr1dOlZ79k= github.com/whyrusleeping/go-keyspace v0.0.0-20160322163242-5b898ac5add1/go.mod h1:8UvriyWtv5Q5EOgjHaSseUEdkQfvwFv1I/In/O2M9gc= github.com/whyrusleeping/go-logging v0.0.0-20170515211332-0457bb6b88fc/go.mod h1:bopw91TMyo8J3tvftk8xmU2kPmlrt4nScJQZU2hE5EM= +github.com/whyrusleeping/go-logging v0.0.1 h1:fwpzlmT0kRC/Fmd0MdmGgJG/CXIZ6gFq46FQZjprUcc= github.com/whyrusleeping/go-logging v0.0.1/go.mod h1:lDPYj54zutzG1XYfHAhcc7oNXEburHQBn+Iqd4yS4vE= github.com/whyrusleeping/go-notifier v0.0.0-20170827234753-097c5d47330f/go.mod h1:cZNvX9cFybI01GriPRMXDtczuvUhgbcYr9iCGaNlRv8= github.com/whyrusleeping/go-smux-multiplex v3.0.16+incompatible/go.mod h1:34LEDbeKFZInPUrAG+bjuJmUXONGdEFW7XL0SpTY1y4= diff --git a/node/builder.go b/node/builder.go index cb3d57721..af0175294 100644 --- a/node/builder.go +++ b/node/builder.go @@ -23,7 +23,7 @@ import ( "github.com/filecoin-project/go-fil-markets/retrievalmarket" "github.com/filecoin-project/go-fil-markets/retrievalmarket/discovery" "github.com/filecoin-project/go-fil-markets/storagemarket" - "github.com/filecoin-project/go-fil-markets/storagemarket/impl/requestvalidation" + "github.com/filecoin-project/go-fil-markets/storagemarket/impl/storedask" "github.com/filecoin-project/specs-actors/actors/runtime" storage2 "github.com/filecoin-project/specs-storage/storage" @@ -259,7 +259,7 @@ func Online() Option { Override(new(dtypes.ClientDealStore), modules.NewClientDealStore), Override(new(dtypes.ClientDatastore), modules.NewClientDatastore), Override(new(dtypes.ClientDataTransfer), modules.NewClientGraphsyncDataTransfer), - Override(new(*requestvalidation.ClientRequestValidator), modules.NewClientRequestValidator), + Override(new(dtypes.ClientRequestValidator), modules.NewClientRequestValidator), Override(new(storagemarket.StorageClient), modules.StorageClient), Override(new(storagemarket.StorageClientNode), storageadapter.NewClientNodeAdapter), Override(RegisterClientValidatorKey, modules.RegisterClientValidator), @@ -300,8 +300,9 @@ func Online() Option { Override(new(retrievalmarket.RetrievalProvider), modules.RetrievalProvider), Override(new(dtypes.ProviderDealStore), modules.NewProviderDealStore), Override(new(dtypes.ProviderDataTransfer), modules.NewProviderDAGServiceDataTransfer), - Override(new(*requestvalidation.ProviderRequestValidator), modules.NewProviderRequestValidator), + Override(new(dtypes.ProviderRequestValidator), modules.NewProviderRequestValidator), Override(new(dtypes.ProviderPieceStore), modules.NewProviderPieceStore), + Override(new(*storedask.StoredAsk), modules.NewStorageAsk), Override(new(storagemarket.StorageProvider), modules.StorageProvider), Override(new(storagemarket.StorageProviderNode), storageadapter.NewProviderNodeAdapter), Override(RegisterProviderValidatorKey, modules.RegisterProviderValidator), diff --git a/node/impl/client/client.go b/node/impl/client/client.go index e1dd0de60..320579ce1 100644 --- a/node/impl/client/client.go +++ b/node/impl/client/client.go @@ -331,7 +331,7 @@ func (a *API) ClientRetrieve(ctx context.Context, order api.RetrievalOrder, ref ppb := types.BigDiv(order.Total, types.NewInt(order.Size)) - a.Retrieval.Retrieve( + _, err := a.Retrieval.Retrieve( ctx, order.Root, retrievalmarket.NewParamsV0(ppb, order.PaymentInterval, order.PaymentIntervalIncrease), @@ -339,6 +339,9 @@ func (a *API) ClientRetrieve(ctx context.Context, order api.RetrievalOrder, ref order.MinerPeerID, order.Client, order.Miner) + if err != nil { + return xerrors.Errorf("Retrieve failed: %w", err) + } select { case <-ctx.Done(): return xerrors.New("Retrieval Timed Out") diff --git a/node/modules/client.go b/node/modules/client.go index 72d80df9a..12ec209b5 100644 --- a/node/modules/client.go +++ b/node/modules/client.go @@ -64,8 +64,8 @@ func ClientBlockstore(fstore dtypes.ClientFilestore) dtypes.ClientBlockstore { // RegisterClientValidator is an initialization hook that registers the client // request validator with the data transfer module as the validator for // StorageDataTransferVoucher types -func RegisterClientValidator(crv *requestvalidation.ClientRequestValidator, dtm dtypes.ClientDataTransfer) { - if err := dtm.RegisterVoucherType(&requestvalidation.StorageDataTransferVoucher{}, crv); err != nil { +func RegisterClientValidator(crv dtypes.ClientRequestValidator, dtm dtypes.ClientDataTransfer) { + if err := dtm.RegisterVoucherType(&requestvalidation.StorageDataTransferVoucher{}, (*requestvalidation.UnifiedRequestValidator)(crv)); err != nil { panic(err) } } @@ -105,8 +105,8 @@ func ClientDAG(mctx helpers.MetricsCtx, lc fx.Lifecycle, ibs dtypes.ClientBlocks return dag } -func NewClientRequestValidator(deals dtypes.ClientDealStore) *requestvalidation.ClientRequestValidator { - return requestvalidation.NewClientRequestValidator(deals) +func NewClientRequestValidator(deals dtypes.ClientDealStore) dtypes.ClientRequestValidator { + return requestvalidation.NewUnifiedRequestValidator(nil, deals) } func StorageClient(lc fx.Lifecycle, h host.Host, ibs dtypes.ClientBlockstore, r repo.LockedRepo, dataTransfer dtypes.ClientDataTransfer, discovery *discovery.Local, deals dtypes.ClientDatastore, scn storagemarket.StorageClientNode) (storagemarket.StorageClient, error) { @@ -133,5 +133,5 @@ func RetrievalClient(h host.Host, bs dtypes.ClientBlockstore, pmgr *paychmgr.Man adapter := retrievaladapter.NewRetrievalClientNode(pmgr, payapi, chainapi) network := rmnet.NewFromLibp2pHost(h) sc := storedcounter.New(ds, datastore.NewKey("/retr")) - return retrievalimpl.NewClient(network, bs, adapter, resolver, ds, sc) + return retrievalimpl.NewClient(network, bs, adapter, resolver, namespace.Wrap(ds, datastore.NewKey("/retrievals/client")), sc) } diff --git a/node/modules/dtypes/storage.go b/node/modules/dtypes/storage.go index 0ec67179e..3eba07e4e 100644 --- a/node/modules/dtypes/storage.go +++ b/node/modules/dtypes/storage.go @@ -1,6 +1,7 @@ package dtypes import ( + "github.com/filecoin-project/go-fil-markets/storagemarket/impl/requestvalidation" bserv "github.com/ipfs/go-blockservice" "github.com/ipfs/go-datastore" "github.com/ipfs/go-filestore" @@ -29,6 +30,7 @@ type ClientFilestore *filestore.Filestore type ClientBlockstore blockstore.Blockstore type ClientDAG format.DAGService type ClientDealStore *statestore.StateStore +type ClientRequestValidator *requestvalidation.UnifiedRequestValidator type ClientDatastore datastore.Batching type Graphsync graphsync.GraphExchange @@ -38,6 +40,7 @@ type ClientDataTransfer datatransfer.Manager type ProviderDealStore *statestore.StateStore type ProviderPieceStore piecestore.PieceStore +type ProviderRequestValidator *requestvalidation.UnifiedRequestValidator // ProviderDataTransfer is a data transfer manager for the provider type ProviderDataTransfer datatransfer.Manager diff --git a/node/modules/services.go b/node/modules/services.go index 7410c5a4f..893ebbf73 100644 --- a/node/modules/services.go +++ b/node/modules/services.go @@ -3,6 +3,8 @@ package modules import ( "context" + "github.com/ipfs/go-datastore" + "github.com/ipfs/go-datastore/namespace" eventbus "github.com/libp2p/go-eventbus" event "github.com/libp2p/go-libp2p-core/event" "github.com/libp2p/go-libp2p-core/host" @@ -114,7 +116,7 @@ func RunDealClient(mctx helpers.MetricsCtx, lc fx.Lifecycle, c storagemarket.Sto } func NewLocalDiscovery(ds dtypes.MetadataDS) *discovery.Local { - return discovery.NewLocal(ds) + return discovery.NewLocal(namespace.Wrap(ds, datastore.NewKey("/deals/local"))) } func RetrievalResolver(l *discovery.Local) retrievalmarket.PeerResolver { diff --git a/node/modules/storageminer.go b/node/modules/storageminer.go index 89fb71d4d..f54a2672e 100644 --- a/node/modules/storageminer.go +++ b/node/modules/storageminer.go @@ -29,11 +29,18 @@ import ( "github.com/filecoin-project/go-fil-markets/storagemarket" storageimpl "github.com/filecoin-project/go-fil-markets/storagemarket/impl" "github.com/filecoin-project/go-fil-markets/storagemarket/impl/requestvalidation" + "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" paramfetch "github.com/filecoin-project/go-paramfetch" "github.com/filecoin-project/go-statestore" "github.com/filecoin-project/go-storedcounter" + sectorstorage "github.com/filecoin-project/sector-storage" + "github.com/filecoin-project/sector-storage/ffiwrapper" + "github.com/filecoin-project/sector-storage/stores" + "github.com/filecoin-project/specs-actors/actors/abi" + sealing "github.com/filecoin-project/storage-fsm" + lapi "github.com/filecoin-project/lotus/api" "github.com/filecoin-project/lotus/build" "github.com/filecoin-project/lotus/chain/beacon" @@ -46,13 +53,10 @@ import ( "github.com/filecoin-project/lotus/node/modules/helpers" "github.com/filecoin-project/lotus/node/repo" "github.com/filecoin-project/lotus/storage" - sectorstorage "github.com/filecoin-project/sector-storage" - "github.com/filecoin-project/sector-storage/ffiwrapper" - "github.com/filecoin-project/sector-storage/stores" - "github.com/filecoin-project/specs-actors/actors/abi" - sealing "github.com/filecoin-project/storage-fsm" ) +var StorageCounterDSPrefix = "/storage/nextid" + func minerAddrFromDS(ds dtypes.MetadataDS) (address.Address, error) { maddrb, err := ds.Get(datastore.NewKey("miner-address")) if err != nil { @@ -117,7 +121,7 @@ func (s *sidsc) Next() (abi.SectorNumber, error) { } func SectorIDCounter(ds dtypes.MetadataDS) sealing.SectorIDCounter { - sc := storedcounter.New(ds, datastore.NewKey("/storage/nextid")) + sc := storedcounter.New(ds, datastore.NewKey(StorageCounterDSPrefix)) return &sidsc{sc} } @@ -191,8 +195,8 @@ func HandleDeals(mctx helpers.MetricsCtx, lc fx.Lifecycle, host host.Host, h sto // RegisterProviderValidator is an initialization hook that registers the provider // request validator with the data transfer module as the validator for // StorageDataTransferVoucher types -func RegisterProviderValidator(mrv *requestvalidation.ProviderRequestValidator, dtm dtypes.ProviderDataTransfer) { - if err := dtm.RegisterVoucherType(&requestvalidation.StorageDataTransferVoucher{}, mrv); err != nil { +func RegisterProviderValidator(mrv dtypes.ProviderRequestValidator, dtm dtypes.ProviderDataTransfer) { + if err := dtm.RegisterVoucherType(&requestvalidation.StorageDataTransferVoucher{}, (*requestvalidation.UnifiedRequestValidator)(mrv)); err != nil { panic(err) } } @@ -206,13 +210,13 @@ func NewProviderDAGServiceDataTransfer(h host.Host, gs dtypes.StagingGraphsync, // NewProviderDealStore creates a statestore for the client to store its deals func NewProviderDealStore(ds dtypes.MetadataDS) dtypes.ProviderDealStore { - return statestore.New(namespace.Wrap(ds, datastore.NewKey("/deals/client"))) + return statestore.New(namespace.Wrap(ds, datastore.NewKey("/deals/provider"))) } // NewProviderPieceStore creates a statestore for storing metadata about pieces // shared by the storage and retrieval providers func NewProviderPieceStore(ds dtypes.MetadataDS) dtypes.ProviderPieceStore { - return piecestore.NewPieceStore(ds) + return piecestore.NewPieceStore(namespace.Wrap(ds, datastore.NewKey("/storagemarket"))) } // StagingBlockstore creates a blockstore for staging blocks for a miner @@ -282,44 +286,35 @@ func SetupBlockProducer(lc fx.Lifecycle, ds dtypes.MetadataDS, api lapi.FullNode return m, nil } -func NewProviderRequestValidator(deals dtypes.ProviderDealStore) *requestvalidation.ProviderRequestValidator { - return requestvalidation.NewProviderRequestValidator(deals) +func NewProviderRequestValidator(deals dtypes.ProviderDealStore) dtypes.ProviderRequestValidator { + return requestvalidation.NewUnifiedRequestValidator(deals, nil) } -func StorageProvider(ctx helpers.MetricsCtx, fapi lapi.FullNode, h host.Host, ds dtypes.MetadataDS, ibs dtypes.StagingBlockstore, r repo.LockedRepo, pieceStore dtypes.ProviderPieceStore, dataTransfer dtypes.ProviderDataTransfer, spn storagemarket.StorageProviderNode) (storagemarket.StorageProvider, error) { +func NewStorageAsk(ctx helpers.MetricsCtx, fapi lapi.FullNode, ds dtypes.MetadataDS, minerAddress dtypes.MinerAddress, spn storagemarket.StorageProviderNode) (*storedask.StoredAsk, error) { + + mi, err := fapi.StateMinerInfo(ctx, address.Address(minerAddress), types.EmptyTSK) + if err != nil { + return nil, err + } + + storedAsk, err := storedask.NewStoredAsk(namespace.Wrap(ds, datastore.NewKey("/deals/provider")), datastore.NewKey("latest-ask"), spn, address.Address(minerAddress)) + // Hacky way to set max piece size to the sector size + a := storedAsk.GetAsk(address.Address(minerAddress)).Ask + err = storedAsk.AddAsk(a.Price, a.Expiry-a.Timestamp, storagemarket.MaxPieceSize(abi.PaddedPieceSize(mi.SectorSize))) + if err != nil { + return storedAsk, err + } + return storedAsk, nil +} + +func StorageProvider(minerAddress dtypes.MinerAddress, ffiConfig *ffiwrapper.Config, storedAsk *storedask.StoredAsk, h host.Host, ds dtypes.MetadataDS, ibs dtypes.StagingBlockstore, r repo.LockedRepo, pieceStore dtypes.ProviderPieceStore, dataTransfer dtypes.ProviderDataTransfer, spn storagemarket.StorageProviderNode) (storagemarket.StorageProvider, error) { + net := smnet.NewFromLibp2pHost(h) store, err := piecefilestore.NewLocalFileStore(piecefilestore.OsPath(r.Path())) if err != nil { return nil, err } - net := smnet.NewFromLibp2pHost(h) - addr, err := ds.Get(datastore.NewKey("miner-address")) - if err != nil { - return nil, err - } - minerAddress, err := address.NewFromBytes(addr) - if err != nil { - return nil, err - } - - mi, err := fapi.StateMinerInfo(ctx, minerAddress, types.EmptyTSK) - if err != nil { - return nil, err - } - - rt, err := ffiwrapper.SealProofTypeFromSectorSize(mi.SectorSize) - if err != nil { - return nil, err - } - - p, err := storageimpl.NewProvider(net, ds, ibs, store, pieceStore, dataTransfer, spn, minerAddress, rt) - if err != nil { - return p, err - } - - // Hacky way to set max piece size to the sector size - a := p.ListAsks(minerAddress)[0].Ask - err = p.AddAsk(a.Price, a.Expiry-a.Timestamp, storagemarket.MaxPieceSize(abi.PaddedPieceSize(mi.SectorSize))) + p, err := storageimpl.NewProvider(net, namespace.Wrap(ds, datastore.NewKey("/deals/provider")), ibs, store, pieceStore, dataTransfer, spn, address.Address(minerAddress), ffiConfig.SealProofType, storedAsk) if err != nil { return p, err } @@ -335,7 +330,7 @@ func RetrievalProvider(h host.Host, miner *storage.Miner, sealer sectorstorage.S return nil, err } network := rmnet.NewFromLibp2pHost(h) - return retrievalimpl.NewProvider(address, adapter, network, pieceStore, ibs, ds) + return retrievalimpl.NewProvider(address, adapter, network, pieceStore, ibs, namespace.Wrap(ds, datastore.NewKey("/retrievals/provider"))) } func SectorStorage(mctx helpers.MetricsCtx, lc fx.Lifecycle, ls stores.LocalStorage, si stores.SectorIndex, cfg *ffiwrapper.Config, sc sectorstorage.SealerConfig, urls sectorstorage.URLs, sa sectorstorage.StorageAuth) (*sectorstorage.Manager, error) { diff --git a/node/node_test.go b/node/node_test.go index bb66b7a39..0b0af0bcf 100644 --- a/node/node_test.go +++ b/node/node_test.go @@ -82,7 +82,7 @@ func testStorageNode(ctx context.Context, t *testing.T, waddr address.Address, a err = ds.Put(datastore.NewKey("miner-address"), act.Bytes()) require.NoError(t, err) - nic := storedcounter.New(ds, datastore.NewKey("/storage/nextid")) + nic := storedcounter.New(ds, datastore.NewKey(modules.StorageCounterDSPrefix)) for i := 0; i < nGenesisPreseals; i++ { nic.Next() }