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

Merge branch 'master' into nonsense/cli-show-deals

Browse Source
This commit is contained in:
Anton Evangelatov
2021-03-23 13:58:26 +02:00
parent ac576c75d4 f145d4f46b
commit 22217b7cae
150 changed files with 6518 additions and 1452 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node/builder.go
+55 -10
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"
@@ -307,9 +308,10 @@ var ChainNode = Options(
Override(new(api.WalletAPI), 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,6 +329,8 @@ 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))),
@@ -334,7 +338,7 @@ var ChainNode = Options(
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(stmgr.StateManagerAPI), rpcstmgr.NewRPCStateManager),
),
// Full node API / service startup
@@ -408,7 +412,7 @@ var MinerNode = Options(
Override(new(dtypes.StorageDealFilter), modules.BasicDealFilter(nil)),
Override(new(storagemarket.StorageProvider), modules.StorageProvider),
Override(new(*storageadapter.DealPublisher), storageadapter.NewDealPublisher(nil, storageadapter.PublishMsgConfig{})),
Override(new(storagemarket.StorageProviderNode), storageadapter.NewProviderNodeAdapter(nil)),
Override(new(storagemarket.StorageProviderNode), storageadapter.NewProviderNodeAdapter(nil, nil)),
Override(HandleMigrateProviderFundsKey, modules.HandleMigrateProviderFunds),
Override(HandleDealsKey, modules.HandleDeals),
@@ -508,6 +512,7 @@ func ConfigCommon(cfg *config.Common) Option {
Override(AddrsFactoryKey, lp2p.AddrsFactory(
cfg.Libp2p.AnnounceAddresses,
cfg.Libp2p.NoAnnounceAddresses)),
Override(new(dtypes.MetadataDS), modules.Datastore(cfg.Backup.DisableMetadataLog)),
)
}
@@ -567,7 +572,7 @@ func ConfigStorageMiner(c interface{}) Option {
Period: time.Duration(cfg.Dealmaking.PublishMsgPeriod),
MaxDealsPerMsg: cfg.Dealmaking.MaxDealsPerPublishMsg,
})),
Override(new(storagemarket.StorageProviderNode), storageadapter.NewProviderNodeAdapter(&cfg.Fees)),
Override(new(storagemarket.StorageProviderNode), storageadapter.NewProviderNodeAdapter(&cfg.Fees, &cfg.Dealmaking)),
Override(new(sectorstorage.SealerConfig), cfg.Storage),
Override(new(*storage.AddressSelector), modules.AddressSelector(&cfg.Addresses)),
@@ -586,14 +591,38 @@ func Repo(r repo.Repo) Option {
return err
}
var cfg *config.Chainstore
switch settings.nodeType {
case repo.FullNode:
cfgp, ok := c.(*config.FullNode)
if !ok {
return xerrors.Errorf("invalid config from repo, got: %T", c)
}
cfg = &cfgp.Chainstore
default:
cfg = &config.Chainstore{}
}
return Options(
Override(new(repo.LockedRepo), modules.LockedRepo(lr)), // module handles closing
Override(new(dtypes.MetadataDS), modules.Datastore),
Override(new(dtypes.UniversalBlockstore), modules.UniversalBlockstore),
Override(new(dtypes.ChainBlockstore), modules.ChainBlockstore),
Override(new(dtypes.StateBlockstore), modules.StateBlockstore),
Override(new(dtypes.ExposedBlockstore), From(new(dtypes.UniversalBlockstore))),
If(cfg.EnableSplitstore,
If(cfg.Splitstore.HotStoreType == "badger",
Override(new(dtypes.HotBlockstore), modules.BadgerHotBlockstore)),
Override(new(dtypes.SplitBlockstore), modules.SplitBlockstore(cfg)),
Override(new(dtypes.ChainBlockstore), modules.ChainSplitBlockstore),
Override(new(dtypes.StateBlockstore), modules.StateSplitBlockstore),
Override(new(dtypes.BaseBlockstore), From(new(dtypes.SplitBlockstore))),
Override(new(dtypes.ExposedBlockstore), From(new(dtypes.SplitBlockstore))),
),
If(!cfg.EnableSplitstore,
Override(new(dtypes.ChainBlockstore), modules.ChainFlatBlockstore),
Override(new(dtypes.StateBlockstore), modules.StateFlatBlockstore),
Override(new(dtypes.BaseBlockstore), From(new(dtypes.UniversalBlockstore))),
Override(new(dtypes.ExposedBlockstore), From(new(dtypes.UniversalBlockstore))),
),
If(os.Getenv("LOTUS_ENABLE_CHAINSTORE_FALLBACK") == "1",
Override(new(dtypes.ChainBlockstore), modules.FallbackChainBlockstore),
@@ -700,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
+52 -10
View File
@@ -13,6 +13,7 @@ import (
// Common is common config between full node and miner
type Common struct {
API API
Backup Backup
Libp2p Libp2p
Pubsub Pubsub
}
@@ -20,14 +21,19 @@ type Common struct {
// FullNode is a full node config
type FullNode struct {
Common
Client Client
Metrics Metrics
Wallet Wallet
Fees FeeConfig
Client Client
Metrics Metrics
Wallet Wallet
Fees FeeConfig
Chainstore Chainstore
}
// // Common
type Backup struct {
DisableMetadataLog bool
}
// StorageMiner is a miner config
type StorageMiner struct {
Common
@@ -54,6 +60,9 @@ type DealmakingConfig struct {
// The maximum number of deals to include in a single PublishStorageDeals
// message
MaxDealsPerPublishMsg uint64
// The maximum collateral that the provider will put up against a deal,
// as a multiplier of the minimum collateral bound
MaxProviderCollateralMultiplier uint64
Filter string
RetrievalFilter string
@@ -91,6 +100,16 @@ type MinerFeeConfig struct {
type MinerAddressConfig struct {
PreCommitControl []string
CommitControl []string
TerminateControl []string
// DisableOwnerFallback disables usage of the owner address for messages
// sent automatically
DisableOwnerFallback bool
// DisableWorkerFallback disables usage of the worker address for messages
// sent automatically, if control addresses are configured.
// A control address that doesn't have enough funds will still be chosen
// over the worker address if this flag is set.
DisableWorkerFallback bool
}
// API contains configs for API endpoint
@@ -114,9 +133,24 @@ type Libp2p struct {
}
type Pubsub struct {
Bootstrapper bool
DirectPeers []string
RemoteTracer string
Bootstrapper bool
DirectPeers []string
IPColocationWhitelist []string
RemoteTracer string
}
type Chainstore struct {
EnableSplitstore bool
Splitstore Splitstore
}
type Splitstore struct {
HotStoreType string
TrackingStoreType string
MarkSetType string
EnableFullCompaction bool
EnableGC bool // EXPERIMENTAL
Archival bool
}
// // Full Node
@@ -184,6 +218,12 @@ func DefaultFullNode() *FullNode {
Client: Client{
SimultaneousTransfers: DefaultSimultaneousTransfers,
},
Chainstore: Chainstore{
EnableSplitstore: false,
Splitstore: Splitstore{
HotStoreType: "badger",
},
},
}
}
@@ -196,6 +236,7 @@ func DefaultStorageMiner() *StorageMiner {
MaxSealingSectors: 0,
MaxSealingSectorsForDeals: 0,
WaitDealsDelay: Duration(time.Hour * 6),
AlwaysKeepUnsealedCopy: true,
},
Storage: sectorstorage.SealerConfig{
@@ -219,9 +260,10 @@ func DefaultStorageMiner() *StorageMiner {
ConsiderUnverifiedStorageDeals: true,
PieceCidBlocklist: []cid.Cid{},
// TODO: It'd be nice to set this based on sector size
ExpectedSealDuration: Duration(time.Hour * 24),
PublishMsgPeriod: Duration(time.Hour),
MaxDealsPerPublishMsg: 8,
ExpectedSealDuration: Duration(time.Hour * 24),
PublishMsgPeriod: Duration(time.Hour),
MaxDealsPerPublishMsg: 8,
MaxProviderCollateralMultiplier: 2,
},
Fees: MinerFeeConfig{
node/impl/common/common.go
+43 -6
View File
@@ -7,6 +7,9 @@ import (
"github.com/gbrlsnchs/jwt/v3"
"github.com/google/uuid"
"go.uber.org/fx"
"golang.org/x/xerrors"
logging "github.com/ipfs/go-log/v2"
"github.com/libp2p/go-libp2p-core/host"
metrics "github.com/libp2p/go-libp2p-core/metrics"
@@ -17,12 +20,11 @@ import (
basichost "github.com/libp2p/go-libp2p/p2p/host/basic"
"github.com/libp2p/go-libp2p/p2p/net/conngater"
ma "github.com/multiformats/go-multiaddr"
"go.uber.org/fx"
"golang.org/x/xerrors"
"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"
@@ -99,6 +101,37 @@ func (a *CommonAPI) NetPeers(context.Context) ([]peer.AddrInfo, error) {
return out, nil
}
func (a *CommonAPI) NetPeerInfo(_ context.Context, p peer.ID) (*api.ExtendedPeerInfo, error) {
info := &api.ExtendedPeerInfo{ID: p}
agent, err := a.Host.Peerstore().Get(p, "AgentVersion")
if err == nil {
info.Agent = agent.(string)
}
for _, a := range a.Host.Peerstore().Addrs(p) {
info.Addrs = append(info.Addrs, a.String())
}
sort.Strings(info.Addrs)
protocols, err := a.Host.Peerstore().GetProtocols(p)
if err == nil {
sort.Strings(protocols)
info.Protocols = protocols
}
if cm := a.Host.ConnManager().GetTagInfo(p); cm != nil {
info.ConnMgrMeta = &api.ConnMgrInfo{
FirstSeen: cm.FirstSeen,
Value: cm.Value,
Tags: cm.Tags,
Conns: cm.Conns,
}
}
return info, nil
}
func (a *CommonAPI) NetConnect(ctx context.Context, p peer.AddrInfo) error {
if swrm, ok := a.Host.Network().(*swarm.Swarm); ok {
swrm.Backoff().Clear(p.ID)
@@ -175,17 +208,21 @@ 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
}
func (a *CommonAPI) Version(context.Context) (api.Version, error) {
v, err := build.VersionForType(build.RunningNodeType)
func (a *CommonAPI) Version(context.Context) (api.APIVersion, error) {
v, err := api.VersionForType(api.RunningNodeType)
if err != nil {
return api.Version{}, err
return api.APIVersion{}, err
}
return api.Version{
return api.APIVersion{
Version: build.UserVersion(),
APIVersion: v,
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/storminer.go
+6
View File
@@ -8,6 +8,7 @@ import (
"strconv"
"time"
"github.com/filecoin-project/lotus/build"
"github.com/google/uuid"
"github.com/ipfs/go-cid"
"github.com/libp2p/go-libp2p-core/host"
@@ -31,6 +32,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"
@@ -690,4 +692,8 @@ 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
}
var _ api.StorageMiner = &StorageMinerAPI{}
node/modules/blockstore.go
+74 -9
View File
@@ -3,19 +3,25 @@ package modules
import (
"context"
"io"
"os"
"path/filepath"
bstore "github.com/ipfs/go-ipfs-blockstore"
"go.uber.org/fx"
"golang.org/x/xerrors"
"github.com/filecoin-project/lotus/blockstore"
badgerbs "github.com/filecoin-project/lotus/blockstore/badger"
"github.com/filecoin-project/lotus/blockstore/splitstore"
"github.com/filecoin-project/lotus/node/config"
"github.com/filecoin-project/lotus/node/modules/dtypes"
"github.com/filecoin-project/lotus/node/modules/helpers"
"github.com/filecoin-project/lotus/node/repo"
)
// UniversalBlockstore returns a single universal blockstore that stores both
// chain data and state data.
// chain data and state data. It can be backed by a blockstore directly
// (e.g. Badger), or by a Splitstore.
func UniversalBlockstore(lc fx.Lifecycle, mctx helpers.MetricsCtx, r repo.LockedRepo) (dtypes.UniversalBlockstore, error) {
bs, err := r.Blockstore(helpers.LifecycleCtx(mctx, lc), repo.UniversalBlockstore)
if err != nil {
@@ -31,17 +37,76 @@ func UniversalBlockstore(lc fx.Lifecycle, mctx helpers.MetricsCtx, r repo.Locked
return bs, err
}
// StateBlockstore is a hook to overlay caches for state objects, or in the
// future, to segregate the universal blockstore into different physical state
// and chain stores.
func StateBlockstore(lc fx.Lifecycle, mctx helpers.MetricsCtx, bs dtypes.UniversalBlockstore) (dtypes.StateBlockstore, error) {
func BadgerHotBlockstore(lc fx.Lifecycle, r repo.LockedRepo) (dtypes.HotBlockstore, error) {
path, err := r.SplitstorePath()
if err != nil {
return nil, err
}
path = filepath.Join(path, "hot.badger")
if err := os.MkdirAll(path, 0755); err != nil {
return nil, err
}
opts, err := repo.BadgerBlockstoreOptions(repo.HotBlockstore, path, r.Readonly())
if err != nil {
return nil, err
}
bs, err := badgerbs.Open(opts)
if err != nil {
return nil, err
}
lc.Append(fx.Hook{
OnStop: func(_ context.Context) error {
return bs.Close()
}})
return bs, nil
}
// ChainBlockstore is a hook to overlay caches for state objects, or in the
// future, to segregate the universal blockstore into different physical state
// and chain stores.
func ChainBlockstore(lc fx.Lifecycle, mctx helpers.MetricsCtx, bs dtypes.UniversalBlockstore) (dtypes.ChainBlockstore, error) {
func SplitBlockstore(cfg *config.Chainstore) func(lc fx.Lifecycle, r repo.LockedRepo, ds dtypes.MetadataDS, cold dtypes.UniversalBlockstore, hot dtypes.HotBlockstore) (dtypes.SplitBlockstore, error) {
return func(lc fx.Lifecycle, r repo.LockedRepo, ds dtypes.MetadataDS, cold dtypes.UniversalBlockstore, hot dtypes.HotBlockstore) (dtypes.SplitBlockstore, error) {
path, err := r.SplitstorePath()
if err != nil {
return nil, err
}
cfg := &splitstore.Config{
TrackingStoreType: cfg.Splitstore.TrackingStoreType,
MarkSetType: cfg.Splitstore.MarkSetType,
EnableFullCompaction: cfg.Splitstore.EnableFullCompaction,
EnableGC: cfg.Splitstore.EnableGC,
Archival: cfg.Splitstore.Archival,
}
ss, err := splitstore.Open(path, ds, hot, cold, cfg)
if err != nil {
return nil, err
}
lc.Append(fx.Hook{
OnStop: func(context.Context) error {
return ss.Close()
},
})
return ss, err
}
}
func StateFlatBlockstore(_ fx.Lifecycle, _ helpers.MetricsCtx, bs dtypes.UniversalBlockstore) (dtypes.StateBlockstore, error) {
return bs, nil
}
func StateSplitBlockstore(_ fx.Lifecycle, _ helpers.MetricsCtx, bs dtypes.SplitBlockstore) (dtypes.StateBlockstore, error) {
return bs, nil
}
func ChainFlatBlockstore(_ fx.Lifecycle, _ helpers.MetricsCtx, bs dtypes.UniversalBlockstore) (dtypes.ChainBlockstore, error) {
return bs, nil
}
func ChainSplitBlockstore(_ fx.Lifecycle, _ helpers.MetricsCtx, bs dtypes.SplitBlockstore) (dtypes.ChainBlockstore, error) {
return bs, nil
}
node/modules/chain.go
+14 -1
View File
@@ -14,6 +14,7 @@ import (
"golang.org/x/xerrors"
"github.com/filecoin-project/lotus/blockstore"
"github.com/filecoin-project/lotus/blockstore/splitstore"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain"
"github.com/filecoin-project/lotus/chain/beacon"
@@ -72,14 +73,26 @@ func MessagePool(lc fx.Lifecycle, sm *stmgr.StateManager, ps *pubsub.PubSub, ds
return mp, nil
}
func ChainStore(lc fx.Lifecycle, cbs dtypes.ChainBlockstore, sbs dtypes.StateBlockstore, ds dtypes.MetadataDS, syscalls vm.SyscallBuilder, j journal.Journal) *store.ChainStore {
func ChainStore(lc fx.Lifecycle, cbs dtypes.ChainBlockstore, sbs dtypes.StateBlockstore, ds dtypes.MetadataDS, basebs dtypes.BaseBlockstore, syscalls vm.SyscallBuilder, j journal.Journal) *store.ChainStore {
chain := store.NewChainStore(cbs, sbs, ds, syscalls, j)
if err := chain.Load(); err != nil {
log.Warnf("loading chain state from disk: %s", err)
}
var startHook func(context.Context) error
if ss, ok := basebs.(*splitstore.SplitStore); ok {
startHook = func(_ context.Context) error {
err := ss.Start(chain)
if err != nil {
err = xerrors.Errorf("error starting splitstore: %w", err)
}
return err
}
}
lc.Append(fx.Hook{
OnStart: startHook,
OnStop: func(_ context.Context) error {
return chain.Close()
},
node/modules/dtypes/storage.go
+9
View File
@@ -27,6 +27,15 @@ type (
// UniversalBlockstore is the cold blockstore.
UniversalBlockstore blockstore.Blockstore
// HotBlockstore is the Hot blockstore abstraction for the splitstore
HotBlockstore blockstore.Blockstore
// SplitBlockstore is the hot/cold blockstore that sits on top of the ColdBlockstore.
SplitBlockstore blockstore.Blockstore
// BaseBlockstore is something, coz DI
BaseBlockstore blockstore.Blockstore
// ChainBlockstore is a blockstore to store chain data (tipsets, blocks,
// messages). It is physically backed by the BareMonolithBlockstore, but it
// has a cache on top that is specially tuned for chain data access
node/modules/lp2p/pubsub.go
+12 -2
View File
@@ -3,6 +3,7 @@ package lp2p
import (
"context"
"encoding/json"
"net"
"time"
host "github.com/libp2p/go-libp2p-core/host"
@@ -198,6 +199,16 @@ func GossipSub(in GossipIn) (service *pubsub.PubSub, err error) {
drandTopics = append(drandTopics, topic)
}
// IP colocation whitelist
var ipcoloWhitelist []*net.IPNet
for _, cidr := range in.Cfg.IPColocationWhitelist {
_, ipnet, err := net.ParseCIDR(cidr)
if err != nil {
return nil, xerrors.Errorf("error parsing IPColocation subnet %s: %w", cidr, err)
}
ipcoloWhitelist = append(ipcoloWhitelist, ipnet)
}
options := []pubsub.Option{
// Gossipsubv1.1 configuration
pubsub.WithFloodPublish(true),
@@ -228,8 +239,7 @@ func GossipSub(in GossipIn) (service *pubsub.PubSub, err error) {
// This sets the IP colocation threshold to 5 peers before we apply penalties
IPColocationFactorThreshold: 5,
IPColocationFactorWeight: -100,
// TODO we want to whitelist IPv6 /64s that belong to datacenters etc
// IPColocationFactorWhitelist: map[string]struct{}{},
IPColocationFactorWhitelist: ipcoloWhitelist,
// P7: behavioural penalties, decay after 1hr
BehaviourPenaltyThreshold: 6,
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/storage.go
+27 -7
View File
@@ -2,8 +2,10 @@ package modules
import (
"context"
"path/filepath"
"go.uber.org/fx"
"golang.org/x/xerrors"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/lib/backupds"
@@ -28,12 +30,30 @@ func KeyStore(lr repo.LockedRepo) (types.KeyStore, error) {
return lr.KeyStore()
}
func Datastore(lc fx.Lifecycle, mctx helpers.MetricsCtx, r repo.LockedRepo) (dtypes.MetadataDS, error) {
ctx := helpers.LifecycleCtx(mctx, lc)
mds, err := r.Datastore(ctx, "/metadata")
if err != nil {
return nil, err
}
func Datastore(disableLog bool) func(lc fx.Lifecycle, mctx helpers.MetricsCtx, r repo.LockedRepo) (dtypes.MetadataDS, error) {
return func(lc fx.Lifecycle, mctx helpers.MetricsCtx, r repo.LockedRepo) (dtypes.MetadataDS, error) {
ctx := helpers.LifecycleCtx(mctx, lc)
mds, err := r.Datastore(ctx, "/metadata")
if err != nil {
return nil, err
}
return backupds.Wrap(mds), nil
var logdir string
if !disableLog {
logdir = filepath.Join(r.Path(), "kvlog/metadata")
}
bds, err := backupds.Wrap(mds, logdir)
if err != nil {
return nil, xerrors.Errorf("opening backupds: %w", err)
}
lc.Append(fx.Hook{
OnStop: func(_ context.Context) error {
return bds.CloseLog()
},
})
return bds, nil
}
}
node/modules/storageminer.go
+12
View File
@@ -157,6 +157,9 @@ func AddressSelector(addrConf *config.MinerAddressConfig) func() (*storage.Addre
return as, nil
}
as.DisableOwnerFallback = addrConf.DisableOwnerFallback
as.DisableWorkerFallback = addrConf.DisableWorkerFallback
for _, s := range addrConf.PreCommitControl {
addr, err := address.NewFromString(s)
if err != nil {
@@ -175,6 +178,15 @@ func AddressSelector(addrConf *config.MinerAddressConfig) func() (*storage.Addre
as.CommitControl = append(as.CommitControl, addr)
}
for _, s := range addrConf.TerminateControl {
addr, err := address.NewFromString(s)
if err != nil {
return nil, xerrors.Errorf("parsing terminate control address: %w", err)
}
as.TerminateControl = append(as.TerminateControl, addr)
}
return as, nil
}
}
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)
node/repo/interface.go
+1
View File
@@ -23,6 +23,7 @@ const (
// well as state. In the future, they may get segregated into different
// domains.
UniversalBlockstore = BlockstoreDomain("universal")
HotBlockstore = BlockstoreDomain("hot")
)
var (