cerc-io/lotus
11
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 27 Wiki Activity

Merge remote-tracking branch 'origin/master' into feat/stateless-offline-dealflow

Browse Source
This commit is contained in:
Peter Rabbitson
2021-04-05 16:06:44 +02:00
parent 7fddbb528d eede19fb0b
commit b79be2a2c2
135 changed files with 5102 additions and 2274 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node/builder.go
+29 -9
View File
@@ -11,6 +11,7 @@ import (
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/lotus/chain"
"github.com/filecoin-project/lotus/chain/exchange"
rpcstmgr "github.com/filecoin-project/lotus/chain/stmgr/rpc"
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/vm"
"github.com/filecoin-project/lotus/chain/wallet"
@@ -304,12 +305,13 @@ var ChainNode = Options(
Override(new(*messagesigner.MessageSigner), messagesigner.NewMessageSigner),
Override(new(*wallet.LocalWallet), wallet.NewWallet),
Override(new(wallet.Default), From(new(*wallet.LocalWallet))),
Override(new(api.WalletAPI), From(new(wallet.MultiWallet))),
Override(new(api.Wallet), From(new(wallet.MultiWallet))),
// Service: Payment channels
Override(new(*paychmgr.Store), paychmgr.NewStore),
Override(new(*paychmgr.Manager), paychmgr.NewManager),
Override(HandlePaymentChannelManagerKey, paychmgr.HandleManager),
Override(new(paychmgr.PaychAPI), From(new(modules.PaychAPI))),
Override(new(*paychmgr.Store), modules.NewPaychStore),
Override(new(*paychmgr.Manager), modules.NewManager),
Override(HandlePaymentChannelManagerKey, modules.HandlePaychManager),
Override(SettlePaymentChannelsKey, settler.SettlePaymentChannels),
// Markets (common)
@@ -327,14 +329,16 @@ var ChainNode = Options(
Override(new(storagemarket.StorageClientNode), storageadapter.NewClientNodeAdapter),
Override(HandleMigrateClientFundsKey, modules.HandleMigrateClientFunds),
Override(new(*full.GasPriceCache), full.NewGasPriceCache),
// Lite node API
ApplyIf(isLiteNode,
Override(new(messagesigner.MpoolNonceAPI), From(new(modules.MpoolNonceAPI))),
Override(new(full.ChainModuleAPI), From(new(api.GatewayAPI))),
Override(new(full.GasModuleAPI), From(new(api.GatewayAPI))),
Override(new(full.MpoolModuleAPI), From(new(api.GatewayAPI))),
Override(new(full.StateModuleAPI), From(new(api.GatewayAPI))),
Override(new(stmgr.StateManagerAPI), modules.NewRPCStateManager),
Override(new(full.ChainModuleAPI), From(new(api.Gateway))),
Override(new(full.GasModuleAPI), From(new(api.Gateway))),
Override(new(full.MpoolModuleAPI), From(new(api.Gateway))),
Override(new(full.StateModuleAPI), From(new(api.Gateway))),
Override(new(stmgr.StateManagerAPI), rpcstmgr.NewRPCStateManager),
),
// Full node API / service startup
@@ -725,3 +729,19 @@ func Test() Option {
Override(new(*storageadapter.DealPublisher), storageadapter.NewDealPublisher(nil, storageadapter.PublishMsgConfig{})),
)
}
// For 3rd party dep injection.
func WithRepoType(repoType repo.RepoType) func(s *Settings) error {
return func(s *Settings) error {
s.nodeType = repoType
return nil
}
}
func WithInvokesKey(i invoke, resApi interface{}) func(s *Settings) error {
return func(s *Settings) error {
s.invokes[i] = fx.Populate(resApi)
return nil
}
}
node/config/def.go
+1
View File
@@ -236,6 +236,7 @@ func DefaultStorageMiner() *StorageMiner {
MaxSealingSectors: 0,
MaxSealingSectorsForDeals: 0,
WaitDealsDelay: Duration(time.Hour * 6),
AlwaysKeepUnsealedCopy: true,
},
Storage: sectorstorage.SealerConfig{
node/impl/client/client.go
+124 -66
View File
@@ -61,6 +61,7 @@ import (
"github.com/filecoin-project/lotus/node/impl/paych"
"github.com/filecoin-project/lotus/node/modules/dtypes"
"github.com/filecoin-project/lotus/node/repo/importmgr"
"github.com/filecoin-project/lotus/node/repo/retrievalstoremgr"
)
var DefaultHashFunction = uint64(mh.BLAKE2B_MIN + 31)
@@ -81,6 +82,7 @@ type API struct {
Chain *store.ChainStore
Imports dtypes.ClientImportMgr
Mds dtypes.ClientMultiDstore
CombinedBstore dtypes.ClientBlockstore // TODO: try to remove
RetrievalStoreMgr dtypes.ClientRetrievalStoreManager
@@ -369,10 +371,14 @@ func (a *API) newDealInfo(ctx context.Context, v storagemarket.ClientDeal) api.D
// be not found if it's no longer active
if err == nil {
ch := api.NewDataTransferChannel(a.Host.ID(), state)
ch.Stages = state.Stages()
transferCh = &ch
}
}
return a.newDealInfoWithTransfer(transferCh, v)
di := a.newDealInfoWithTransfer(transferCh, v)
di.DealStages = v.DealStages
return di
}
func (a *API) newDealInfoWithTransfer(transferCh *api.DataTransferChannel, v storagemarket.ClientDeal) api.DealInfo {
@@ -577,6 +583,29 @@ func (a *API) ClientListImports(ctx context.Context) ([]api.Import, error) {
return out, nil
}
func (a *API) ClientCancelRetrievalDeal(ctx context.Context, dealID retrievalmarket.DealID) error {
cerr := make(chan error)
go func() {
err := a.Retrieval.CancelDeal(dealID)
select {
case cerr <- err:
case <-ctx.Done():
}
}()
select {
case err := <-cerr:
if err != nil {
return xerrors.Errorf("failed to cancel retrieval deal: %w", err)
}
return nil
case <-ctx.Done():
return xerrors.Errorf("context timeout while canceling retrieval deal: %w", ctx.Err())
}
}
func (a *API) ClientRetrieve(ctx context.Context, order api.RetrievalOrder, ref *api.FileRef) error {
events := make(chan marketevents.RetrievalEvent)
go a.clientRetrieve(ctx, order, ref, events)
@@ -657,86 +686,102 @@ func (a *API) clientRetrieve(ctx context.Context, order api.RetrievalOrder, ref
}
}
if order.MinerPeer.ID == "" {
mi, err := a.StateMinerInfo(ctx, order.Miner, types.EmptyTSK)
if err != nil {
finish(err)
var store retrievalstoremgr.RetrievalStore
if order.LocalStore == nil {
if order.MinerPeer == nil || order.MinerPeer.ID == "" {
mi, err := a.StateMinerInfo(ctx, order.Miner, types.EmptyTSK)
if err != nil {
finish(err)
return
}
order.MinerPeer = &retrievalmarket.RetrievalPeer{
ID: *mi.PeerId,
Address: order.Miner,
}
}
if order.Size == 0 {
finish(xerrors.Errorf("cannot make retrieval deal for zero bytes"))
return
}
order.MinerPeer = retrievalmarket.RetrievalPeer{
ID: *mi.PeerId,
Address: order.Miner,
/*id, st, err := a.imgr().NewStore()
if err != nil {
return err
}
}
if err := a.imgr().AddLabel(id, "source", "retrieval"); err != nil {
return err
}*/
if order.Size == 0 {
finish(xerrors.Errorf("cannot make retrieval deal for zero bytes"))
return
}
ppb := types.BigDiv(order.Total, types.NewInt(order.Size))
/*id, st, err := a.imgr().NewStore()
if err != nil {
return err
}
if err := a.imgr().AddLabel(id, "source", "retrieval"); err != nil {
return err
}*/
params, err := rm.NewParamsV1(ppb, order.PaymentInterval, order.PaymentIntervalIncrease, shared.AllSelector(), order.Piece, order.UnsealPrice)
if err != nil {
finish(xerrors.Errorf("Error in retrieval params: %s", err))
return
}
ppb := types.BigDiv(order.Total, types.NewInt(order.Size))
store, err = a.RetrievalStoreMgr.NewStore()
if err != nil {
finish(xerrors.Errorf("Error setting up new store: %w", err))
return
}
params, err := rm.NewParamsV1(ppb, order.PaymentInterval, order.PaymentIntervalIncrease, shared.AllSelector(), order.Piece, order.UnsealPrice)
if err != nil {
finish(xerrors.Errorf("Error in retrieval params: %s", err))
return
}
defer func() {
_ = a.RetrievalStoreMgr.ReleaseStore(store)
}()
store, err := a.RetrievalStoreMgr.NewStore()
if err != nil {
finish(xerrors.Errorf("Error setting up new store: %w", err))
return
}
defer func() {
_ = a.RetrievalStoreMgr.ReleaseStore(store)
}()
// Subscribe to events before retrieving to avoid losing events.
subscribeEvents := make(chan retrievalSubscribeEvent, 1)
subscribeCtx, cancel := context.WithCancel(ctx)
defer cancel()
unsubscribe := a.Retrieval.SubscribeToEvents(func(event rm.ClientEvent, state rm.ClientDealState) {
// We'll check the deal IDs inside readSubscribeEvents.
if state.PayloadCID.Equals(order.Root) {
select {
case <-subscribeCtx.Done():
case subscribeEvents <- retrievalSubscribeEvent{event, state}:
// Subscribe to events before retrieving to avoid losing events.
subscribeEvents := make(chan retrievalSubscribeEvent, 1)
subscribeCtx, cancel := context.WithCancel(ctx)
defer cancel()
unsubscribe := a.Retrieval.SubscribeToEvents(func(event rm.ClientEvent, state rm.ClientDealState) {
// We'll check the deal IDs inside readSubscribeEvents.
if state.PayloadCID.Equals(order.Root) {
select {
case <-subscribeCtx.Done():
case subscribeEvents <- retrievalSubscribeEvent{event, state}:
}
}
})
dealID, err := a.Retrieval.Retrieve(
ctx,
order.Root,
params,
order.Total,
*order.MinerPeer,
order.Client,
order.Miner,
store.StoreID())
if err != nil {
unsubscribe()
finish(xerrors.Errorf("Retrieve failed: %w", err))
return
}
})
dealID, err := a.Retrieval.Retrieve(
ctx,
order.Root,
params,
order.Total,
order.MinerPeer,
order.Client,
order.Miner,
store.StoreID())
err = readSubscribeEvents(ctx, dealID, subscribeEvents, events)
if err != nil {
unsubscribe()
finish(xerrors.Errorf("Retrieve failed: %w", err))
return
}
if err != nil {
finish(xerrors.Errorf("Retrieve: %w", err))
return
}
} else {
// local retrieval
st, err := ((*multistore.MultiStore)(a.Mds)).Get(*order.LocalStore)
if err != nil {
finish(xerrors.Errorf("Retrieve: %w", err))
return
}
err = readSubscribeEvents(ctx, dealID, subscribeEvents, events)
unsubscribe()
if err != nil {
finish(xerrors.Errorf("Retrieve: %w", err))
return
store = &multiStoreRetrievalStore{
storeID: *order.LocalStore,
store: st,
}
}
// If ref is nil, it only fetches the data into the configured blockstore.
@@ -776,6 +821,19 @@ func (a *API) clientRetrieve(ctx context.Context, order api.RetrievalOrder, ref
return
}
type multiStoreRetrievalStore struct {
storeID multistore.StoreID
store *multistore.Store
}
func (mrs *multiStoreRetrievalStore) StoreID() *multistore.StoreID {
return &mrs.storeID
}
func (mrs *multiStoreRetrievalStore) DAGService() ipld.DAGService {
return mrs.store.DAG
}
func (a *API) ClientQueryAsk(ctx context.Context, p peer.ID, miner address.Address) (*storagemarket.StorageAsk, error) {
mi, err := a.StateMinerInfo(ctx, miner, types.EmptyTSK)
if err != nil {
node/impl/common/common.go
+5
View File
@@ -24,6 +24,7 @@ import (
"github.com/filecoin-project/go-jsonrpc/auth"
"github.com/filecoin-project/lotus/api"
apitypes "github.com/filecoin-project/lotus/api/types"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/node/modules/dtypes"
"github.com/filecoin-project/lotus/node/modules/lp2p"
@@ -207,6 +208,10 @@ func (a *CommonAPI) NetBandwidthStatsByProtocol(ctx context.Context) (map[protoc
return a.Reporter.GetBandwidthByProtocol(), nil
}
func (a *CommonAPI) Discover(ctx context.Context) (apitypes.OpenRPCDocument, error) {
return build.OpenRPCDiscoverJSON_Full(), nil
}
func (a *CommonAPI) ID(context.Context) (peer.ID, error) {
return a.Host.ID(), nil
}
node/impl/full/gas.go
+61 -22
View File
@@ -8,6 +8,7 @@ import (
"github.com/filecoin-project/lotus/chain/actors/builtin"
"github.com/filecoin-project/lotus/chain/actors/builtin/paych"
lru "github.com/hashicorp/golang-lru"
"go.uber.org/fx"
"golang.org/x/xerrors"
@@ -39,6 +40,8 @@ type GasModule struct {
Chain *store.ChainStore
Mpool *messagepool.MessagePool
GetMaxFee dtypes.DefaultMaxFeeFunc
PriceCache *GasPriceCache
}
var _ GasModuleAPI = (*GasModule)(nil)
@@ -51,6 +54,53 @@ type GasAPI struct {
Stmgr *stmgr.StateManager
Chain *store.ChainStore
Mpool *messagepool.MessagePool
PriceCache *GasPriceCache
}
func NewGasPriceCache() *GasPriceCache {
// 50 because we usually won't access more than 40
c, err := lru.New2Q(50)
if err != nil {
// err only if parameter is bad
panic(err)
}
return &GasPriceCache{
c: c,
}
}
type GasPriceCache struct {
c *lru.TwoQueueCache
}
type GasMeta struct {
Price big.Int
Limit int64
}
func (g *GasPriceCache) GetTSGasStats(cstore *store.ChainStore, ts *types.TipSet) ([]GasMeta, error) {
i, has := g.c.Get(ts.Key())
if has {
return i.([]GasMeta), nil
}
var prices []GasMeta
msgs, err := cstore.MessagesForTipset(ts)
if err != nil {
return nil, xerrors.Errorf("loading messages: %w", err)
}
for _, msg := range msgs {
prices = append(prices, GasMeta{
Price: msg.VMMessage().GasPremium,
Limit: msg.VMMessage().GasLimit,
})
}
g.c.Add(ts.Key(), prices)
return prices, nil
}
const MinGasPremium = 100e3
@@ -88,24 +138,19 @@ func gasEstimateFeeCap(cstore *store.ChainStore, msg *types.Message, maxqueueblk
return out, nil
}
type gasMeta struct {
price big.Int
limit int64
}
// finds 55th percntile instead of median to put negative pressure on gas price
func medianGasPremium(prices []gasMeta, blocks int) abi.TokenAmount {
func medianGasPremium(prices []GasMeta, blocks int) abi.TokenAmount {
sort.Slice(prices, func(i, j int) bool {
// sort desc by price
return prices[i].price.GreaterThan(prices[j].price)
return prices[i].Price.GreaterThan(prices[j].Price)
})
at := build.BlockGasTarget * int64(blocks) / 2 // 50th
at += build.BlockGasTarget * int64(blocks) / (2 * 20) // move 5% further
prev1, prev2 := big.Zero(), big.Zero()
for _, price := range prices {
prev1, prev2 = price.price, prev1
at -= price.limit
prev1, prev2 = price.Price, prev1
at -= price.Limit
if at < 0 {
break
}
@@ -126,7 +171,7 @@ func (a *GasAPI) GasEstimateGasPremium(
gaslimit int64,
_ types.TipSetKey,
) (types.BigInt, error) {
return gasEstimateGasPremium(a.Chain, nblocksincl)
return gasEstimateGasPremium(a.Chain, a.PriceCache, nblocksincl)
}
func (m *GasModule) GasEstimateGasPremium(
ctx context.Context,
@@ -135,14 +180,14 @@ func (m *GasModule) GasEstimateGasPremium(
gaslimit int64,
_ types.TipSetKey,
) (types.BigInt, error) {
return gasEstimateGasPremium(m.Chain, nblocksincl)
return gasEstimateGasPremium(m.Chain, m.PriceCache, nblocksincl)
}
func gasEstimateGasPremium(cstore *store.ChainStore, nblocksincl uint64) (types.BigInt, error) {
func gasEstimateGasPremium(cstore *store.ChainStore, cache *GasPriceCache, nblocksincl uint64) (types.BigInt, error) {
if nblocksincl == 0 {
nblocksincl = 1
}
var prices []gasMeta
var prices []GasMeta
var blocks int
ts := cstore.GetHeaviestTipSet()
@@ -157,17 +202,11 @@ func gasEstimateGasPremium(cstore *store.ChainStore, nblocksincl uint64) (types.
}
blocks += len(pts.Blocks())
msgs, err := cstore.MessagesForTipset(pts)
meta, err := cache.GetTSGasStats(cstore, pts)
if err != nil {
return types.BigInt{}, xerrors.Errorf("loading messages: %w", err)
}
for _, msg := range msgs {
prices = append(prices, gasMeta{
price: msg.VMMessage().GasPremium,
limit: msg.VMMessage().GasLimit,
})
return types.BigInt{}, err
}
prices = append(prices, meta...)
ts = pts
}
node/impl/full/gas_test.go
+5 -5
View File
@@ -12,27 +12,27 @@ import (
)
func TestMedian(t *testing.T) {
require.Equal(t, types.NewInt(5), medianGasPremium([]gasMeta{
require.Equal(t, types.NewInt(5), medianGasPremium([]GasMeta{
{big.NewInt(5), build.BlockGasTarget},
}, 1))
require.Equal(t, types.NewInt(10), medianGasPremium([]gasMeta{
require.Equal(t, types.NewInt(10), medianGasPremium([]GasMeta{
{big.NewInt(5), build.BlockGasTarget},
{big.NewInt(10), build.BlockGasTarget},
}, 1))
require.Equal(t, types.NewInt(15), medianGasPremium([]gasMeta{
require.Equal(t, types.NewInt(15), medianGasPremium([]GasMeta{
{big.NewInt(10), build.BlockGasTarget / 2},
{big.NewInt(20), build.BlockGasTarget / 2},
}, 1))
require.Equal(t, types.NewInt(25), medianGasPremium([]gasMeta{
require.Equal(t, types.NewInt(25), medianGasPremium([]GasMeta{
{big.NewInt(10), build.BlockGasTarget / 2},
{big.NewInt(20), build.BlockGasTarget / 2},
{big.NewInt(30), build.BlockGasTarget / 2},
}, 1))
require.Equal(t, types.NewInt(15), medianGasPremium([]gasMeta{
require.Equal(t, types.NewInt(15), medianGasPremium([]GasMeta{
{big.NewInt(10), build.BlockGasTarget / 2},
{big.NewInt(20), build.BlockGasTarget / 2},
{big.NewInt(30), build.BlockGasTarget / 2},
node/impl/full/state.go
+1 -1
View File
@@ -76,7 +76,7 @@ type StateAPI struct {
// TODO: the wallet here is only needed because we have the MinerCreateBlock
// API attached to the state API. It probably should live somewhere better
Wallet api.WalletAPI
Wallet api.Wallet
DefWallet wallet.Default
StateModuleAPI
node/impl/full/wallet.go
+3 -3
View File
@@ -22,7 +22,7 @@ type WalletAPI struct {
StateManagerAPI stmgr.StateManagerAPI
Default wallet.Default
api.WalletAPI
api.Wallet
}
func (a *WalletAPI) WalletBalance(ctx context.Context, addr address.Address) (types.BigInt, error) {
@@ -40,7 +40,7 @@ func (a *WalletAPI) WalletSign(ctx context.Context, k address.Address, msg []byt
if err != nil {
return nil, xerrors.Errorf("failed to resolve ID address: %w", keyAddr)
}
return a.WalletAPI.WalletSign(ctx, keyAddr, msg, api.MsgMeta{
return a.Wallet.WalletSign(ctx, keyAddr, msg, api.MsgMeta{
Type: api.MTUnknown,
})
}
@@ -56,7 +56,7 @@ func (a *WalletAPI) WalletSignMessage(ctx context.Context, k address.Address, ms
return nil, xerrors.Errorf("serializing message: %w", err)
}
sig, err := a.WalletAPI.WalletSign(ctx, keyAddr, mb.Cid().Bytes(), api.MsgMeta{
sig, err := a.Wallet.WalletSign(ctx, keyAddr, mb.Cid().Bytes(), api.MsgMeta{
Type: api.MTChainMsg,
Extra: mb.RawData(),
})
node/impl/remoteworker.go
+1 -1
View File
@@ -16,7 +16,7 @@ import (
)
type remoteWorker struct {
api.WorkerAPI
api.Worker
closer jsonrpc.ClientCloser
}
node/impl/storminer.go
+15 -1
View File
@@ -8,6 +8,10 @@ import (
"strconv"
"time"
"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"
@@ -31,6 +35,7 @@ import (
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/api/apistruct"
apitypes "github.com/filecoin-project/lotus/api/types"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/markets/storageadapter"
"github.com/filecoin-project/lotus/miner"
@@ -61,7 +66,8 @@ type StorageMinerAPI struct {
AddrSel *storage.AddressSelector
DealPublisher *storageadapter.DealPublisher
DS dtypes.MetadataDS
Epp gen.WinningPoStProver
DS dtypes.MetadataDS
ConsiderOnlineStorageDealsConfigFunc dtypes.ConsiderOnlineStorageDealsConfigFunc
SetConsiderOnlineStorageDealsConfigFunc dtypes.SetConsiderOnlineStorageDealsConfigFunc
@@ -690,4 +696,12 @@ func (sm *StorageMinerAPI) ActorAddressConfig(ctx context.Context) (api.AddressC
return sm.AddrSel.AddressConfig, nil
}
func (sm *StorageMinerAPI) Discover(ctx context.Context) (apitypes.OpenRPCDocument, error) {
return build.OpenRPCDiscoverJSON_Miner(), nil
}
func (sm *StorageMinerAPI) ComputeProof(ctx context.Context, ssi []builtin.SectorInfo, rand abi.PoStRandomness) ([]builtin.PoStProof, error) {
return sm.Epp.ComputeProof(ctx, ssi, rand)
}
var _ api.StorageMiner = &StorageMinerAPI{}
node/modules/client.go
+28 -9
View File
@@ -7,12 +7,10 @@ import (
"path/filepath"
"time"
"github.com/filecoin-project/go-multistore"
"github.com/filecoin-project/go-state-types/abi"
"go.uber.org/fx"
"golang.org/x/xerrors"
"go.uber.org/fx"
"github.com/filecoin-project/go-data-transfer/channelmonitor"
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"
@@ -25,6 +23,8 @@ import (
storageimpl "github.com/filecoin-project/go-fil-markets/storagemarket/impl"
"github.com/filecoin-project/go-fil-markets/storagemarket/impl/requestvalidation"
smnet "github.com/filecoin-project/go-fil-markets/storagemarket/network"
"github.com/filecoin-project/go-multistore"
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/go-storedcounter"
"github.com/ipfs/go-datastore"
"github.com/ipfs/go-datastore/namespace"
@@ -121,8 +121,6 @@ func RegisterClientValidator(crv dtypes.ClientRequestValidator, dtm dtypes.Clien
// NewClientGraphsyncDataTransfer returns a data transfer manager that just
// uses the clients's Client DAG service for transfers
func NewClientGraphsyncDataTransfer(lc fx.Lifecycle, h host.Host, gs dtypes.Graphsync, ds dtypes.MetadataDS, r repo.LockedRepo) (dtypes.ClientDataTransfer, error) {
sc := storedcounter.New(ds, datastore.NewKey("/datatransfer/client/counter"))
// go-data-transfer protocol retries:
// 1s, 5s, 25s, 2m5s, 5m x 11 ~= 1 hour
dtRetryParams := dtnet.RetryParameters(time.Second, 5*time.Minute, 15, 5)
@@ -135,9 +133,30 @@ func NewClientGraphsyncDataTransfer(lc fx.Lifecycle, h host.Host, gs dtypes.Grap
return nil, err
}
// data-transfer push channel restart configuration
dtRestartConfig := dtimpl.PushChannelRestartConfig(time.Minute, 10, 1024, 10*time.Minute, 3)
dt, err := dtimpl.NewDataTransfer(dtDs, filepath.Join(r.Path(), "data-transfer"), net, transport, sc, dtRestartConfig)
// data-transfer push / pull channel restart configuration:
dtRestartConfig := dtimpl.ChannelRestartConfig(channelmonitor.Config{
// For now only monitor push channels (for storage deals)
MonitorPushChannels: true,
// TODO: Enable pull channel monitoring (for retrievals) when the
// following issue has been fixed:
// https://github.com/filecoin-project/go-data-transfer/issues/172
MonitorPullChannels: false,
// Wait up to 30s for the other side to respond to an Open channel message
AcceptTimeout: 30 * time.Second,
// Send a restart message if the data rate falls below 1024 bytes / minute
Interval: time.Minute,
MinBytesTransferred: 1024,
// Perform check 10 times / minute
ChecksPerInterval: 10,
// After sending a restart, wait for at least 1 minute before sending another
RestartBackoff: time.Minute,
// After trying to restart 3 times, give up and fail the transfer
MaxConsecutiveRestarts: 3,
// Wait up to 30s for the other side to send a Complete message once all
// data has been sent / received
CompleteTimeout: 30 * time.Second,
})
dt, err := dtimpl.NewDataTransfer(dtDs, filepath.Join(r.Path(), "data-transfer"), net, transport, dtRestartConfig)
if err != nil {
return nil, err
}
node/modules/paych.go
+45
View File
@@ -0,0 +1,45 @@
package modules
import (
"context"
"github.com/filecoin-project/lotus/chain/stmgr"
"github.com/filecoin-project/lotus/node/impl/full"
"github.com/filecoin-project/lotus/node/modules/dtypes"
"github.com/filecoin-project/lotus/node/modules/helpers"
"github.com/filecoin-project/lotus/paychmgr"
"github.com/ipfs/go-datastore"
"github.com/ipfs/go-datastore/namespace"
"go.uber.org/fx"
)
func NewManager(mctx helpers.MetricsCtx, lc fx.Lifecycle, sm stmgr.StateManagerAPI, pchstore *paychmgr.Store, api paychmgr.PaychAPI) *paychmgr.Manager {
ctx := helpers.LifecycleCtx(mctx, lc)
ctx, shutdown := context.WithCancel(ctx)
return paychmgr.NewManager(ctx, shutdown, sm, pchstore, api)
}
func NewPaychStore(ds dtypes.MetadataDS) *paychmgr.Store {
ds = namespace.Wrap(ds, datastore.NewKey("/paych/"))
return paychmgr.NewStore(ds)
}
type PaychAPI struct {
fx.In
full.MpoolAPI
full.StateAPI
}
// HandlePaychManager is called by dependency injection to set up hooks
func HandlePaychManager(lc fx.Lifecycle, pm *paychmgr.Manager) {
lc.Append(fx.Hook{
OnStart: func(ctx context.Context) error {
return pm.Start()
},
OnStop: func(context.Context) error {
return pm.Stop()
},
})
}
node/modules/rpcstatemanager.go
-58
View File
@@ -1,58 +0,0 @@
package modules
import (
"context"
"golang.org/x/xerrors"
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/blockstore"
"github.com/filecoin-project/lotus/chain/actors/adt"
"github.com/filecoin-project/lotus/chain/actors/builtin/paych"
"github.com/filecoin-project/lotus/chain/stmgr"
"github.com/filecoin-project/lotus/chain/types"
cbor "github.com/ipfs/go-ipld-cbor"
)
type RPCStateManager struct {
gapi api.GatewayAPI
cstore *cbor.BasicIpldStore
}
func NewRPCStateManager(api api.GatewayAPI) *RPCStateManager {
cstore := cbor.NewCborStore(blockstore.NewAPIBlockstore(api))
return &RPCStateManager{gapi: api, cstore: cstore}
}
func (s *RPCStateManager) GetPaychState(ctx context.Context, addr address.Address, ts *types.TipSet) (*types.Actor, paych.State, error) {
act, err := s.gapi.StateGetActor(ctx, addr, ts.Key())
if err != nil {
return nil, nil, err
}
actState, err := paych.Load(adt.WrapStore(ctx, s.cstore), act)
if err != nil {
return nil, nil, err
}
return act, actState, nil
}
func (s *RPCStateManager) LoadActorTsk(ctx context.Context, addr address.Address, tsk types.TipSetKey) (*types.Actor, error) {
return s.gapi.StateGetActor(ctx, addr, tsk)
}
func (s *RPCStateManager) LookupID(ctx context.Context, addr address.Address, ts *types.TipSet) (address.Address, error) {
return s.gapi.StateLookupID(ctx, addr, ts.Key())
}
func (s *RPCStateManager) ResolveToKeyAddress(ctx context.Context, addr address.Address, ts *types.TipSet) (address.Address, error) {
return s.gapi.StateAccountKey(ctx, addr, ts.Key())
}
func (s *RPCStateManager) Call(ctx context.Context, msg *types.Message, ts *types.TipSet) (*api.InvocResult, error) {
return nil, xerrors.Errorf("RPCStateManager does not implement StateManager.Call")
}
var _ stmgr.StateManagerAPI = (*RPCStateManager)(nil)
node/modules/storageminer.go
+1 -2
View File
@@ -330,7 +330,6 @@ func HandleMigrateProviderFunds(lc fx.Lifecycle, ds dtypes.MetadataDS, node api.
// NewProviderDAGServiceDataTransfer returns a data transfer manager that just
// uses the provider's Staging DAG service for transfers
func NewProviderDAGServiceDataTransfer(lc fx.Lifecycle, h host.Host, gs dtypes.StagingGraphsync, ds dtypes.MetadataDS, r repo.LockedRepo) (dtypes.ProviderDataTransfer, error) {
sc := storedcounter.New(ds, datastore.NewKey("/datatransfer/provider/counter"))
net := dtnet.NewFromLibp2pHost(h)
dtDs := namespace.Wrap(ds, datastore.NewKey("/datatransfer/provider/transfers"))
@@ -340,7 +339,7 @@ func NewProviderDAGServiceDataTransfer(lc fx.Lifecycle, h host.Host, gs dtypes.S
return nil, err
}
dt, err := dtimpl.NewDataTransfer(dtDs, filepath.Join(r.Path(), "data-transfer"), net, transport, sc)
dt, err := dtimpl.NewDataTransfer(dtDs, filepath.Join(r.Path(), "data-transfer"), net, transport)
if err != nil {
return nil, err
}
node/node_test.go
+3
View File
@@ -58,6 +58,9 @@ func TestAPIDealFlow(t *testing.T) {
t.Run("TestPublishDealsBatching", func(t *testing.T) {
test.TestPublishDealsBatching(t, builder.MockSbBuilder, blockTime, dealStartEpoch)
})
t.Run("TestBatchDealInput", func(t *testing.T) {
test.TestBatchDealInput(t, builder.MockSbBuilder, blockTime, dealStartEpoch)
})
}
func TestAPIDealFlowReal(t *testing.T) {
node/repo/fsrepo.go
+15 -1
View File
@@ -545,17 +545,31 @@ func (fsr *fsLockedRepo) Get(name string) (types.KeyInfo, error) {
return res, nil
}
const KTrashPrefix = "trash-"
// Put saves key info under given name
func (fsr *fsLockedRepo) Put(name string, info types.KeyInfo) error {
return fsr.put(name, info, 0)
}
func (fsr *fsLockedRepo) put(rawName string, info types.KeyInfo, retries int) error {
if err := fsr.stillValid(); err != nil {
return err
}
name := rawName
if retries > 0 {
name = fmt.Sprintf("%s-%d", rawName, retries)
}
encName := base32.RawStdEncoding.EncodeToString([]byte(name))
keyPath := fsr.join(fsKeystore, encName)
_, err := os.Stat(keyPath)
if err == nil {
if err == nil && strings.HasPrefix(name, KTrashPrefix) {
// retry writing the trash-prefixed file with a number suffix
return fsr.put(rawName, info, retries+1)
} else if err == nil {
return xerrors.Errorf("checking key before put '%s': %w", name, types.ErrKeyExists)
} else if !os.IsNotExist(err) {
return xerrors.Errorf("checking key before put '%s': %w", name, err)