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

Merge remote-tracking branch 'origin/master' into 3692-config-default-update

Browse Source
This commit is contained in:
Łukasz Magiera
2020-10-02 13:33:17 +02:00
parent c610bd818e efc1b24f4e
commit a572bef8b2
318 changed files with 18632 additions and 7672 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node/builder.go
+54 -19
View File
@@ -3,6 +3,7 @@ package node
import (
"context"
"errors"
"os"
"time"
logging "github.com/ipfs/go-log"
@@ -19,8 +20,9 @@ import (
"go.uber.org/fx"
"golang.org/x/xerrors"
"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/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/storedask"
@@ -34,6 +36,7 @@ import (
"github.com/filecoin-project/lotus/chain/gen/slashfilter"
"github.com/filecoin-project/lotus/chain/market"
"github.com/filecoin-project/lotus/chain/messagepool"
"github.com/filecoin-project/lotus/chain/messagesigner"
"github.com/filecoin-project/lotus/chain/metrics"
"github.com/filecoin-project/lotus/chain/stmgr"
"github.com/filecoin-project/lotus/chain/store"
@@ -44,6 +47,7 @@ import (
"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/journal"
"github.com/filecoin-project/lotus/lib/blockstore"
"github.com/filecoin-project/lotus/lib/peermgr"
_ "github.com/filecoin-project/lotus/lib/sigs/bls"
@@ -67,6 +71,10 @@ import (
"github.com/filecoin-project/lotus/storage/sectorblocks"
)
// EnvJournalDisabledEvents is the environment variable through which disabled
// journal events can be customized.
const EnvJournalDisabledEvents = "LOTUS_JOURNAL_DISABLED_EVENTS"
//nolint:deadcode,varcheck
var log = logging.Logger("builder")
@@ -91,11 +99,16 @@ var (
type invoke int
// Invokes are called in the order they are defined.
//nolint:golint
const (
// InitJournal at position 0 initializes the journal global var as soon as
// the system starts, so that it's available for all other components.
InitJournalKey = invoke(iota)
// libp2p
PstoreAddSelfKeysKey = invoke(iota)
PstoreAddSelfKeysKey
StartListeningKey
BootstrapKey
@@ -123,7 +136,6 @@ const (
HeadMetricsKey
SettlePaymentChannelsKey
RunPeerTaggerKey
JournalKey
SetApiEndpointKey
@@ -151,11 +163,25 @@ type Settings struct {
func defaults() []Option {
return []Option{
// global system journal.
Override(new(journal.DisabledEvents), func() journal.DisabledEvents {
if env, ok := os.LookupEnv(EnvJournalDisabledEvents); ok {
if ret, err := journal.ParseDisabledEvents(env); err == nil {
return ret
}
}
// fallback if env variable is not set, or if it failed to parse.
return journal.DefaultDisabledEvents
}),
Override(new(journal.Journal), modules.OpenFilesystemJournal),
Override(InitJournalKey, func(j journal.Journal) {
journal.J = j // eagerly sets the global journal through fx.Invoke.
}),
Override(new(helpers.MetricsCtx), context.Background),
Override(new(record.Validator), modules.RecordValidator),
Override(new(dtypes.Bootstrapper), dtypes.Bootstrapper(false)),
Override(new(dtypes.ShutdownChan), make(chan struct{})),
Override(JournalKey, modules.SetupJournal),
// Filecoin modules
@@ -226,7 +252,7 @@ func Online() Option {
Override(new(dtypes.BootstrapPeers), modules.BuiltinBootstrap),
Override(new(dtypes.DrandBootstrap), modules.DrandBootstrap),
Override(new(dtypes.DrandConfig), modules.BuiltinDrandConfig),
Override(new(dtypes.DrandSchedule), modules.BuiltinDrandConfig),
Override(HandleIncomingMessagesKey, modules.HandleIncomingMessages),
@@ -235,6 +261,7 @@ func Online() Option {
Override(new(*store.ChainStore), modules.ChainStore),
Override(new(*stmgr.StateManager), stmgr.NewStateManager),
Override(new(*wallet.Wallet), wallet.NewWallet),
Override(new(*messagesigner.MessageSigner), messagesigner.NewMessageSigner),
Override(new(dtypes.ChainGCLocker), blockstore.NewGCLocker),
Override(new(dtypes.ChainGCBlockstore), modules.ChainGCBlockstore),
@@ -242,6 +269,9 @@ func Online() Option {
Override(new(dtypes.ChainBlockService), modules.ChainBlockService),
// Filecoin services
// We don't want the SyncManagerCtor to be used as an fx constructor, but rather as a value.
// It will be called implicitly by the Syncer constructor.
Override(new(chain.SyncManagerCtor), func() chain.SyncManagerCtor { return chain.NewSyncManager }),
Override(new(*chain.Syncer), modules.NewSyncer),
Override(new(exchange.Client), exchange.NewClient),
Override(new(*messagepool.MessagePool), modules.MessagePool),
@@ -263,8 +293,8 @@ func Online() Option {
Override(RunPeerMgrKey, modules.RunPeerMgr),
Override(HandleIncomingBlocksKey, modules.HandleIncomingBlocks),
Override(new(*discovery.Local), modules.NewLocalDiscovery),
Override(new(retrievalmarket.PeerResolver), modules.RetrievalResolver),
Override(new(*discoveryimpl.Local), modules.NewLocalDiscovery),
Override(new(discovery.PeerResolver), modules.RetrievalResolver),
Override(new(retrievalmarket.RetrievalClient), modules.RetrievalClient),
Override(new(dtypes.ClientDatastore), modules.NewClientDatastore),
@@ -272,7 +302,7 @@ func Online() Option {
Override(new(modules.ClientDealFunds), modules.NewClientDealFunds),
Override(new(storagemarket.StorageClient), modules.StorageClient),
Override(new(storagemarket.StorageClientNode), storageadapter.NewClientNodeAdapter),
Override(new(beacon.RandomBeacon), modules.RandomBeacon),
Override(new(beacon.Schedule), modules.RandomSchedule),
Override(new(*paychmgr.Store), paychmgr.NewStore),
Override(new(*paychmgr.Manager), paychmgr.NewManager),
@@ -355,7 +385,7 @@ func StorageMiner(out *api.StorageMiner) Option {
func(s *Settings) error {
resAPI := &impl.StorageMinerAPI{}
s.invokes[ExtractApiKey] = fx.Extract(resAPI)
s.invokes[ExtractApiKey] = fx.Populate(resAPI)
*out = resAPI
return nil
},
@@ -475,12 +505,18 @@ func Repo(r repo.Repo) Option {
}
func FullAPI(out *api.FullNode) Option {
return func(s *Settings) error {
resAPI := &impl.FullNodeAPI{}
s.invokes[ExtractApiKey] = fx.Extract(resAPI)
*out = resAPI
return nil
}
return Options(
func(s *Settings) error {
s.nodeType = repo.FullNode
return nil
},
func(s *Settings) error {
resAPI := &impl.FullNodeAPI{}
s.invokes[ExtractApiKey] = fx.Populate(resAPI)
*out = resAPI
return nil
},
)
}
type StopFunc func(context.Context) error
@@ -488,9 +524,8 @@ type StopFunc func(context.Context) error
// New builds and starts new Filecoin node
func New(ctx context.Context, opts ...Option) (StopFunc, error) {
settings := Settings{
modules: map[interface{}]fx.Option{},
invokes: make([]fx.Option, _nInvokes),
nodeType: repo.FullNode,
modules: map[interface{}]fx.Option{},
invokes: make([]fx.Option, _nInvokes),
}
// apply module options in the right order
@@ -535,6 +570,6 @@ func Test() Option {
return Options(
Unset(RunPeerMgrKey),
Unset(new(*peermgr.PeerMgr)),
Override(new(beacon.RandomBeacon), testing.RandomBeacon),
Override(new(beacon.Schedule), testing.RandomBeacon),
)
}
node/impl/client/client.go
+11 -8
View File
@@ -6,6 +6,8 @@ import (
"io"
"os"
"github.com/filecoin-project/go-state-types/dline"
datatransfer "github.com/filecoin-project/go-data-transfer"
"github.com/filecoin-project/go-state-types/big"
"golang.org/x/xerrors"
@@ -31,6 +33,7 @@ import (
"go.uber.org/fx"
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/go-fil-markets/discovery"
"github.com/filecoin-project/go-fil-markets/pieceio"
"github.com/filecoin-project/go-fil-markets/retrievalmarket"
rm "github.com/filecoin-project/go-fil-markets/retrievalmarket"
@@ -39,7 +42,7 @@ import (
"github.com/filecoin-project/go-multistore"
"github.com/filecoin-project/go-padreader"
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/specs-actors/actors/builtin/miner"
miner0 "github.com/filecoin-project/specs-actors/actors/builtin/miner"
"github.com/filecoin-project/lotus/extern/sector-storage/ffiwrapper"
marketevents "github.com/filecoin-project/lotus/markets/loggers"
@@ -68,7 +71,7 @@ type API struct {
paych.PaychAPI
SMDealClient storagemarket.StorageClient
RetDiscovery rm.PeerResolver
RetDiscovery discovery.PeerResolver
Retrieval rm.RetrievalClient
Chain *store.ChainStore
@@ -80,12 +83,12 @@ type API struct {
Host host.Host
}
func calcDealExpiration(minDuration uint64, md *miner.DeadlineInfo, startEpoch abi.ChainEpoch) abi.ChainEpoch {
func calcDealExpiration(minDuration uint64, md *dline.Info, startEpoch abi.ChainEpoch) abi.ChainEpoch {
// Make sure we give some time for the miner to seal
minExp := startEpoch + abi.ChainEpoch(minDuration)
// Align on miners ProvingPeriodBoundary
return minExp + miner.WPoStProvingPeriod - (minExp % miner.WPoStProvingPeriod) + (md.PeriodStart % miner.WPoStProvingPeriod) - 1
return minExp + miner0.WPoStProvingPeriod - (minExp % miner0.WPoStProvingPeriod) + (md.PeriodStart % miner0.WPoStProvingPeriod) - 1
}
func (a *API) imgr() *importmgr.Mgr {
@@ -612,18 +615,18 @@ func (a *API) clientRetrieve(ctx context.Context, order api.RetrievalOrder, ref
return
}
func (a *API) ClientQueryAsk(ctx context.Context, p peer.ID, miner address.Address) (*storagemarket.SignedStorageAsk, error) {
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 {
return nil, xerrors.Errorf("failed getting miner info: %w", err)
}
info := utils.NewStorageProviderInfo(miner, mi.Worker, mi.SectorSize, p, mi.Multiaddrs)
signedAsk, err := a.SMDealClient.GetAsk(ctx, info)
ask, err := a.SMDealClient.GetAsk(ctx, info)
if err != nil {
return nil, err
}
return signedAsk, nil
return ask, nil
}
func (a *API) ClientCalcCommP(ctx context.Context, inpath string) (*api.CommPRet, error) {
@@ -709,7 +712,7 @@ func (a *API) ClientGenCar(ctx context.Context, ref api.FileRef, outputPath stri
// TODO: does that defer mean to remove the whole blockstore?
defer bufferedDS.Remove(ctx, c) //nolint:errcheck
ssb := builder.NewSelectorSpecBuilder(basicnode.Style.Any)
ssb := builder.NewSelectorSpecBuilder(basicnode.Prototype.Any)
// entire DAG selector
allSelector := ssb.ExploreRecursive(selector.RecursionLimitNone(),
node/impl/common/common.go
+6 -1
View File
@@ -170,9 +170,14 @@ func (a *CommonAPI) ID(context.Context) (peer.ID, error) {
}
func (a *CommonAPI) Version(context.Context) (api.Version, error) {
v, err := build.VersionForType(build.RunningNodeType)
if err != nil {
return api.Version{}, err
}
return api.Version{
Version: build.UserVersion(),
APIVersion: build.APIVersion,
APIVersion: v,
BlockDelay: build.BlockDelaySecs,
}, nil
node/impl/full/beacon.go
+4 -3
View File
@@ -13,12 +13,13 @@ import (
type BeaconAPI struct {
fx.In
Beacon beacon.RandomBeacon
Beacon beacon.Schedule
}
func (a *BeaconAPI) BeaconGetEntry(ctx context.Context, epoch abi.ChainEpoch) (*types.BeaconEntry, error) {
rr := a.Beacon.MaxBeaconRoundForEpoch(epoch, types.BeaconEntry{})
e := a.Beacon.Entry(ctx, rr)
b := a.Beacon.BeaconForEpoch(epoch)
rr := b.MaxBeaconRoundForEpoch(epoch)
e := b.Entry(ctx, rr)
select {
case be, ok := <-e:
node/impl/full/chain.go
+8 -4
View File
@@ -197,6 +197,10 @@ func (a *ChainAPI) ChainReadObj(ctx context.Context, obj cid.Cid) ([]byte, error
return blk.RawData(), nil
}
func (a *ChainAPI) ChainDeleteObj(ctx context.Context, obj cid.Cid) error {
return a.Chain.Blockstore().DeleteBlock(obj)
}
func (a *ChainAPI) ChainHasObj(ctx context.Context, obj cid.Cid) (bool, error) {
return a.Chain.Blockstore().Has(obj)
}
@@ -300,7 +304,7 @@ func resolveOnce(bs blockstore.Blockstore) func(ctx context.Context, ds ipld.Nod
return nil, nil, xerrors.Errorf("parsing int64: %w", err)
}
ik := adt.IntKey(i)
ik := abi.IntKey(i)
names[0] = "@H:" + ik.Key()
}
@@ -311,7 +315,7 @@ func resolveOnce(bs blockstore.Blockstore) func(ctx context.Context, ds ipld.Nod
return nil, nil, xerrors.Errorf("parsing uint64: %w", err)
}
ik := adt.UIntKey(i)
ik := abi.UIntKey(i)
names[0] = "@H:" + ik.Key()
}
@@ -495,7 +499,7 @@ func (a *ChainAPI) ChainGetMessage(ctx context.Context, mc cid.Cid) (*types.Mess
return cm.VMMessage(), nil
}
func (a *ChainAPI) ChainExport(ctx context.Context, nroots abi.ChainEpoch, tsk types.TipSetKey) (<-chan []byte, error) {
func (a *ChainAPI) ChainExport(ctx context.Context, nroots abi.ChainEpoch, skipoldmsgs bool, tsk types.TipSetKey) (<-chan []byte, error) {
ts, err := a.Chain.GetTipSetFromKey(tsk)
if err != nil {
return nil, xerrors.Errorf("loading tipset %s: %w", tsk, err)
@@ -508,7 +512,7 @@ func (a *ChainAPI) ChainExport(ctx context.Context, nroots abi.ChainEpoch, tsk t
bw := bufio.NewWriterSize(w, 1<<20)
defer bw.Flush() //nolint:errcheck // it is a write to a pipe
if err := a.Chain.Export(ctx, ts, nroots, bw); err != nil {
if err := a.Chain.Export(ctx, ts, nroots, skipoldmsgs, bw); err != nil {
log.Errorf("chain export call failed: %s", err)
return
}
node/impl/full/gas.go
+14 -23
View File
@@ -13,7 +13,8 @@ import (
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/go-state-types/big"
"github.com/filecoin-project/go-state-types/exitcode"
"github.com/filecoin-project/specs-actors/actors/builtin"
builtin0 "github.com/filecoin-project/specs-actors/actors/builtin"
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/build"
@@ -166,8 +167,14 @@ func (a *GasAPI) GasEstimateGasLimit(ctx context.Context, msgIn *types.Message,
}
// Special case for PaymentChannel collect, which is deleting actor
var act types.Actor
err = a.Stmgr.WithParentState(ts, a.Stmgr.WithActor(msg.To, stmgr.GetActor(&act)))
st, err := a.Stmgr.ParentState(ts)
if err != nil {
_ = err
// somewhat ignore it as it can happen and we just want to detect
// an existing PaymentChannel actor
return res.MsgRct.GasUsed, nil
}
act, err := st.GetActor(msg.To)
if err != nil {
_ = err
// somewhat ignore it as it can happen and we just want to detect
@@ -175,10 +182,10 @@ func (a *GasAPI) GasEstimateGasLimit(ctx context.Context, msgIn *types.Message,
return res.MsgRct.GasUsed, nil
}
if !act.Code.Equals(builtin.PaymentChannelActorCodeID) {
if !act.IsPaymentChannelActor() {
return res.MsgRct.GasUsed, nil
}
if msgIn.Method != builtin.MethodsPaych.Collect {
if msgIn.Method != builtin0.MethodsPaych.Collect {
return res.MsgRct.GasUsed, nil
}
@@ -204,30 +211,14 @@ func (a *GasAPI) GasEstimateMessageGas(ctx context.Context, msg *types.Message,
}
if msg.GasFeeCap == types.EmptyInt || types.BigCmp(msg.GasFeeCap, types.NewInt(0)) == 0 {
feeCap, err := a.GasEstimateFeeCap(ctx, msg, 10, types.EmptyTSK)
feeCap, err := a.GasEstimateFeeCap(ctx, msg, 20, types.EmptyTSK)
if err != nil {
return nil, xerrors.Errorf("estimating fee cap: %w", err)
}
msg.GasFeeCap = feeCap
}
capGasFee(msg, spec.Get().MaxFee)
messagepool.CapGasFee(msg, spec.Get().MaxFee)
return msg, nil
}
func capGasFee(msg *types.Message, maxFee abi.TokenAmount) {
if maxFee.Equals(big.Zero()) {
maxFee = types.NewInt(build.FilecoinPrecision / 10)
}
gl := types.NewInt(uint64(msg.GasLimit))
totalFee := types.BigMul(msg.GasFeeCap, gl)
if totalFee.LessThanEqual(maxFee) {
return
}
msg.GasFeeCap = big.Div(maxFee, gl)
msg.GasPremium = big.Min(msg.GasFeeCap, msg.GasPremium) // cap premium at FeeCap
}
node/impl/full/mpool.go
+31 -34
View File
@@ -10,8 +10,7 @@ import (
"golang.org/x/xerrors"
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/chain/messagepool"
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/messagesigner"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/node/modules/dtypes"
)
@@ -22,9 +21,7 @@ type MpoolAPI struct {
WalletAPI
GasAPI
Chain *store.ChainStore
Mpool *messagepool.MessagePool
MessageSigner *messagesigner.MessageSigner
PushLocks *dtypes.MpoolLocker
}
@@ -114,12 +111,14 @@ func (a *MpoolAPI) MpoolPush(ctx context.Context, smsg *types.SignedMessage) (ci
}
func (a *MpoolAPI) MpoolPushMessage(ctx context.Context, msg *types.Message, spec *api.MessageSendSpec) (*types.SignedMessage, error) {
cp := *msg
msg = &cp
inMsg := *msg
fromA, err := a.Stmgr.ResolveToKeyAddress(ctx, msg.From, nil)
if err != nil {
return nil, xerrors.Errorf("getting key address: %w", err)
}
{
fromA, err := a.Stmgr.ResolveToKeyAddress(ctx, msg.From, nil)
if err != nil {
return nil, xerrors.Errorf("getting key address: %w", err)
}
done, err := a.PushLocks.TakeLock(ctx, fromA)
if err != nil {
return nil, xerrors.Errorf("taking lock: %w", err)
@@ -131,7 +130,7 @@ func (a *MpoolAPI) MpoolPushMessage(ctx context.Context, msg *types.Message, spe
return nil, xerrors.Errorf("MpoolPushMessage expects message nonce to be 0, was %d", msg.Nonce)
}
msg, err := a.GasAPI.GasEstimateMessageGas(ctx, msg, spec, types.EmptyTSK)
msg, err = a.GasAPI.GasEstimateMessageGas(ctx, msg, spec, types.EmptyTSK)
if err != nil {
return nil, xerrors.Errorf("GasEstimateMessageGas error: %w", err)
}
@@ -143,33 +142,31 @@ func (a *MpoolAPI) MpoolPushMessage(ctx context.Context, msg *types.Message, spe
inJson, outJson)
}
sign := func(from address.Address, nonce uint64) (*types.SignedMessage, error) {
msg.Nonce = nonce
if msg.From.Protocol() == address.ID {
log.Warnf("Push from ID address (%s), adjusting to %s", msg.From, from)
msg.From = from
}
b, err := a.WalletBalance(ctx, msg.From)
if err != nil {
return nil, xerrors.Errorf("mpool push: getting origin balance: %w", err)
}
if b.LessThan(msg.Value) {
return nil, xerrors.Errorf("mpool push: not enough funds: %s < %s", b, msg.Value)
}
return a.WalletSignMessage(ctx, from, msg)
if msg.From.Protocol() == address.ID {
log.Warnf("Push from ID address (%s), adjusting to %s", msg.From, fromA)
msg.From = fromA
}
var m *types.SignedMessage
again:
m, err = a.Mpool.PushWithNonce(ctx, msg.From, sign)
if err == messagepool.ErrTryAgain {
log.Debugf("temporary failure while pushing message: %s; retrying", err)
goto again
b, err := a.WalletBalance(ctx, msg.From)
if err != nil {
return nil, xerrors.Errorf("mpool push: getting origin balance: %w", err)
}
return m, err
if b.LessThan(msg.Value) {
return nil, xerrors.Errorf("mpool push: not enough funds: %s < %s", b, msg.Value)
}
smsg, err := a.MessageSigner.SignMessage(ctx, msg)
if err != nil {
return nil, xerrors.Errorf("mpool push: failed to sign message: %w", err)
}
_, err = a.Mpool.Push(smsg)
if err != nil {
return nil, xerrors.Errorf("mpool push: failed to push message: %w", err)
}
return smsg, err
}
func (a *MpoolAPI) MpoolGetNonce(ctx context.Context, addr address.Address) (uint64, error) {
node/impl/full/multisig.go
+61 -19
View File
@@ -9,10 +9,12 @@ import (
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/chain/actors"
init_ "github.com/filecoin-project/lotus/chain/actors/builtin/init"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/specs-actors/actors/builtin"
init_ "github.com/filecoin-project/specs-actors/actors/builtin/init"
samsig "github.com/filecoin-project/specs-actors/actors/builtin/multisig"
builtin0 "github.com/filecoin-project/specs-actors/actors/builtin"
init0 "github.com/filecoin-project/specs-actors/actors/builtin/init"
multisig0 "github.com/filecoin-project/specs-actors/actors/builtin/multisig"
"github.com/ipfs/go-cid"
"github.com/minio/blake2b-simd"
@@ -46,7 +48,7 @@ func (a *MsigAPI) MsigCreate(ctx context.Context, req uint64, addrs []address.Ad
}
// Set up constructor parameters for multisig
msigParams := &samsig.ConstructorParams{
msigParams := &multisig0.ConstructorParams{
Signers: addrs,
NumApprovalsThreshold: req,
UnlockDuration: duration,
@@ -58,8 +60,9 @@ func (a *MsigAPI) MsigCreate(ctx context.Context, req uint64, addrs []address.Ad
}
// new actors are created by invoking 'exec' on the init actor with the constructor params
execParams := &init_.ExecParams{
CodeCID: builtin.MultisigActorCodeID,
// TODO: network upgrade?
execParams := &init0.ExecParams{
CodeCID: builtin0.MultisigActorCodeID,
ConstructorParams: enc,
}
@@ -70,9 +73,9 @@ func (a *MsigAPI) MsigCreate(ctx context.Context, req uint64, addrs []address.Ad
// now we create the message to send this with
msg := types.Message{
To: builtin.InitActorAddr,
To: init_.Address,
From: src,
Method: builtin.MethodsInit.Exec,
Method: builtin0.MethodsInit.Exec,
Params: enc,
Value: val,
}
@@ -104,7 +107,7 @@ func (a *MsigAPI) MsigPropose(ctx context.Context, msig address.Address, to addr
return cid.Undef, xerrors.Errorf("must provide source address")
}
enc, actErr := actors.SerializeParams(&samsig.ProposeParams{
enc, actErr := actors.SerializeParams(&multisig0.ProposeParams{
To: to,
Value: amt,
Method: abi.MethodNum(method),
@@ -118,7 +121,7 @@ func (a *MsigAPI) MsigPropose(ctx context.Context, msig address.Address, to addr
To: msig,
From: src,
Value: types.NewInt(0),
Method: builtin.MethodsMultisig.Propose,
Method: builtin0.MethodsMultisig.Propose,
Params: enc,
}
@@ -130,13 +133,40 @@ func (a *MsigAPI) MsigPropose(ctx context.Context, msig address.Address, to addr
return smsg.Cid(), nil
}
func (a *MsigAPI) MsigAddPropose(ctx context.Context, msig address.Address, src address.Address, newAdd address.Address, inc bool) (cid.Cid, error) {
enc, actErr := serializeAddParams(newAdd, inc)
if actErr != nil {
return cid.Undef, actErr
}
return a.MsigPropose(ctx, msig, msig, big.Zero(), src, uint64(builtin0.MethodsMultisig.AddSigner), enc)
}
func (a *MsigAPI) MsigAddApprove(ctx context.Context, msig address.Address, src address.Address, txID uint64, proposer address.Address, newAdd address.Address, inc bool) (cid.Cid, error) {
enc, actErr := serializeAddParams(newAdd, inc)
if actErr != nil {
return cid.Undef, actErr
}
return a.MsigApprove(ctx, msig, txID, proposer, msig, big.Zero(), src, uint64(builtin0.MethodsMultisig.AddSigner), enc)
}
func (a *MsigAPI) MsigAddCancel(ctx context.Context, msig address.Address, src address.Address, txID uint64, newAdd address.Address, inc bool) (cid.Cid, error) {
enc, actErr := serializeAddParams(newAdd, inc)
if actErr != nil {
return cid.Undef, actErr
}
return a.MsigCancel(ctx, msig, txID, msig, big.Zero(), src, uint64(builtin0.MethodsMultisig.AddSigner), enc)
}
func (a *MsigAPI) MsigSwapPropose(ctx context.Context, msig address.Address, src address.Address, oldAdd address.Address, newAdd address.Address) (cid.Cid, error) {
enc, actErr := serializeSwapParams(oldAdd, newAdd)
if actErr != nil {
return cid.Undef, actErr
}
return a.MsigPropose(ctx, msig, msig, big.Zero(), src, uint64(builtin.MethodsMultisig.SwapSigner), enc)
return a.MsigPropose(ctx, msig, msig, big.Zero(), src, uint64(builtin0.MethodsMultisig.SwapSigner), enc)
}
func (a *MsigAPI) MsigSwapApprove(ctx context.Context, msig address.Address, src address.Address, txID uint64, proposer address.Address, oldAdd address.Address, newAdd address.Address) (cid.Cid, error) {
@@ -145,7 +175,7 @@ func (a *MsigAPI) MsigSwapApprove(ctx context.Context, msig address.Address, src
return cid.Undef, actErr
}
return a.MsigApprove(ctx, msig, txID, proposer, msig, big.Zero(), src, uint64(builtin.MethodsMultisig.SwapSigner), enc)
return a.MsigApprove(ctx, msig, txID, proposer, msig, big.Zero(), src, uint64(builtin0.MethodsMultisig.SwapSigner), enc)
}
func (a *MsigAPI) MsigSwapCancel(ctx context.Context, msig address.Address, src address.Address, txID uint64, oldAdd address.Address, newAdd address.Address) (cid.Cid, error) {
@@ -154,7 +184,7 @@ func (a *MsigAPI) MsigSwapCancel(ctx context.Context, msig address.Address, src
return cid.Undef, actErr
}
return a.MsigCancel(ctx, msig, txID, msig, big.Zero(), src, uint64(builtin.MethodsMultisig.SwapSigner), enc)
return a.MsigCancel(ctx, msig, txID, msig, big.Zero(), src, uint64(builtin0.MethodsMultisig.SwapSigner), enc)
}
func (a *MsigAPI) MsigApprove(ctx context.Context, msig address.Address, txID uint64, proposer address.Address, to address.Address, amt types.BigInt, src address.Address, method uint64, params []byte) (cid.Cid, error) {
@@ -190,7 +220,7 @@ func (a *MsigAPI) msigApproveOrCancel(ctx context.Context, operation api.MsigPro
proposer = proposerID
}
p := samsig.ProposalHashData{
p := multisig0.ProposalHashData{
Requester: proposer,
To: to,
Value: amt,
@@ -204,8 +234,8 @@ func (a *MsigAPI) msigApproveOrCancel(ctx context.Context, operation api.MsigPro
}
phash := blake2b.Sum256(pser)
enc, err := actors.SerializeParams(&samsig.TxnIDParams{
ID: samsig.TxnID(txID),
enc, err := actors.SerializeParams(&multisig0.TxnIDParams{
ID: multisig0.TxnID(txID),
ProposalHash: phash[:],
})
@@ -221,9 +251,9 @@ func (a *MsigAPI) msigApproveOrCancel(ctx context.Context, operation api.MsigPro
*/
switch operation {
case api.MsigApprove:
msigResponseMethod = builtin.MethodsMultisig.Approve
msigResponseMethod = builtin0.MethodsMultisig.Approve
case api.MsigCancel:
msigResponseMethod = builtin.MethodsMultisig.Cancel
msigResponseMethod = builtin0.MethodsMultisig.Cancel
default:
return cid.Undef, xerrors.Errorf("Invalid operation for msigApproveOrCancel")
}
@@ -244,8 +274,20 @@ func (a *MsigAPI) msigApproveOrCancel(ctx context.Context, operation api.MsigPro
return smsg.Cid(), nil
}
func serializeAddParams(new address.Address, inc bool) ([]byte, error) {
enc, actErr := actors.SerializeParams(&multisig0.AddSignerParams{
Signer: new,
Increase: inc,
})
if actErr != nil {
return nil, actErr
}
return enc, nil
}
func serializeSwapParams(old address.Address, new address.Address) ([]byte, error) {
enc, actErr := actors.SerializeParams(&samsig.SwapSignerParams{
enc, actErr := actors.SerializeParams(&multisig0.SwapSignerParams{
From: old,
To: new,
})
node/impl/full/state.go
+506 -507
View File
File diff suppressed because it is too large Load Diff
node/impl/full/sync.go
+24 -1
View File
@@ -2,6 +2,7 @@ package full
import (
"context"
"sync/atomic"
cid "github.com/ipfs/go-cid"
pubsub "github.com/libp2p/go-libp2p-pubsub"
@@ -13,6 +14,7 @@ import (
"github.com/filecoin-project/lotus/chain"
"github.com/filecoin-project/lotus/chain/gen/slashfilter"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/chain/vm"
"github.com/filecoin-project/lotus/node/modules/dtypes"
)
@@ -28,7 +30,9 @@ type SyncAPI struct {
func (a *SyncAPI) SyncState(ctx context.Context) (*api.SyncState, error) {
states := a.Syncer.State()
out := &api.SyncState{}
out := &api.SyncState{
VMApplied: atomic.LoadUint64(&vm.StatApplied),
}
for i := range states {
ss := &states[i]
@@ -122,3 +126,22 @@ func (a *SyncAPI) SyncCheckBad(ctx context.Context, bcid cid.Cid) (string, error
return reason, nil
}
func (a *SyncAPI) SyncValidateTipset(ctx context.Context, tsk types.TipSetKey) (bool, error) {
ts, err := a.Syncer.ChainStore().LoadTipSet(tsk)
if err != nil {
return false, err
}
fts, err := a.Syncer.ChainStore().TryFillTipSet(ts)
if err != nil {
return false, err
}
err = a.Syncer.ValidateTipSet(ctx, fts, false)
if err != nil {
return false, err
}
return true, nil
}
node/impl/full/wallet.go
+10 -9
View File
@@ -36,16 +36,13 @@ func (a *WalletAPI) WalletList(ctx context.Context) ([]address.Address, error) {
}
func (a *WalletAPI) WalletBalance(ctx context.Context, addr address.Address) (types.BigInt, error) {
var bal types.BigInt
err := a.StateManager.WithParentStateTsk(types.EmptyTSK, a.StateManager.WithActor(addr, func(act *types.Actor) error {
bal = act.Balance
return nil
}))
act, err := a.StateManager.LoadActorTsk(ctx, addr, types.EmptyTSK)
if xerrors.Is(err, types.ErrActorNotFound) {
return big.Zero(), nil
} else if err != nil {
return big.Zero(), err
}
return bal, err
return act.Balance, nil
}
func (a *WalletAPI) WalletSign(ctx context.Context, k address.Address, msg []byte) (*crypto.Signature, error) {
@@ -70,8 +67,8 @@ func (a *WalletAPI) WalletSignMessage(ctx context.Context, k address.Address, ms
}, nil
}
func (a *WalletAPI) WalletVerify(ctx context.Context, k address.Address, msg []byte, sig *crypto.Signature) bool {
return sigs.Verify(sig, k, msg) == nil
func (a *WalletAPI) WalletVerify(ctx context.Context, k address.Address, msg []byte, sig *crypto.Signature) (bool, error) {
return sigs.Verify(sig, k, msg) == nil, nil
}
func (a *WalletAPI) WalletDefaultAddress(ctx context.Context) (address.Address, error) {
@@ -93,3 +90,7 @@ func (a *WalletAPI) WalletImport(ctx context.Context, ki *types.KeyInfo) (addres
func (a *WalletAPI) WalletDelete(ctx context.Context, addr address.Address) error {
return a.Wallet.DeleteKey(addr)
}
func (a *WalletAPI) WalletValidateAddress(ctx context.Context, str string) (address.Address, error) {
return address.NewFromString(str)
}
node/impl/paych/paych.go
+3 -3
View File
@@ -9,9 +9,9 @@ import (
"go.uber.org/fx"
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/specs-actors/actors/builtin/paych"
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/chain/actors/builtin/paych"
"github.com/filecoin-project/lotus/chain/types"
full "github.com/filecoin-project/lotus/node/impl/full"
"github.com/filecoin-project/lotus/paychmgr"
@@ -39,11 +39,11 @@ func (a *PaychAPI) PaychGet(ctx context.Context, from, to address.Address, amt t
}, nil
}
func (a *PaychAPI) PaychAvailableFunds(ch address.Address) (*api.ChannelAvailableFunds, error) {
func (a *PaychAPI) PaychAvailableFunds(ctx context.Context, ch address.Address) (*api.ChannelAvailableFunds, error) {
return a.PaychMgr.AvailableFunds(ch)
}
func (a *PaychAPI) PaychAvailableFundsByFromTo(from, to address.Address) (*api.ChannelAvailableFunds, error) {
func (a *PaychAPI) PaychAvailableFundsByFromTo(ctx context.Context, from, to address.Address) (*api.ChannelAvailableFunds, error) {
return a.PaychMgr.AvailableFundsByFromTo(from, to)
}
node/impl/storminer.go
+26 -4
View File
@@ -305,8 +305,30 @@ func (sm *StorageMinerAPI) MarketImportDealData(ctx context.Context, propCid cid
return sm.StorageProvider.ImportDataForDeal(ctx, propCid, fi)
}
func (sm *StorageMinerAPI) MarketListDeals(ctx context.Context) ([]storagemarket.StorageDeal, error) {
return sm.StorageProvider.ListDeals(ctx)
func (sm *StorageMinerAPI) listDeals(ctx context.Context) ([]api.MarketDeal, error) {
ts, err := sm.Full.ChainHead(ctx)
if err != nil {
return nil, err
}
tsk := ts.Key()
allDeals, err := sm.Full.StateMarketDeals(ctx, tsk)
if err != nil {
return nil, err
}
var out []api.MarketDeal
for _, deal := range allDeals {
if deal.Proposal.Provider == sm.Miner.Address() {
out = append(out, deal)
}
}
return out, nil
}
func (sm *StorageMinerAPI) MarketListDeals(ctx context.Context) ([]api.MarketDeal, error) {
return sm.listDeals(ctx)
}
func (sm *StorageMinerAPI) MarketListRetrievalDeals(ctx context.Context) ([]retrievalmarket.ProviderDealState, error) {
@@ -395,8 +417,8 @@ func (sm *StorageMinerAPI) MarketDataTransferUpdates(ctx context.Context) (<-cha
return channels, nil
}
func (sm *StorageMinerAPI) DealsList(ctx context.Context) ([]storagemarket.StorageDeal, error) {
return sm.StorageProvider.ListDeals(ctx)
func (sm *StorageMinerAPI) DealsList(ctx context.Context) ([]api.MarketDeal, error) {
return sm.listDeals(ctx)
}
func (sm *StorageMinerAPI) RetrievalDealsList(ctx context.Context) (map[retrievalmarket.ProviderDealIdentifier]retrievalmarket.ProviderDealState, error) {
node/modules/chain.go
+30 -2
View File
@@ -18,6 +18,7 @@ import (
"github.com/filecoin-project/lotus/extern/sector-storage/ffiwrapper"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain"
"github.com/filecoin-project/lotus/chain/beacon"
"github.com/filecoin-project/lotus/chain/exchange"
@@ -157,14 +158,41 @@ func SetGenesis(cs *store.ChainStore, g Genesis) (dtypes.AfterGenesisSet, error)
}
func NetworkName(mctx helpers.MetricsCtx, lc fx.Lifecycle, cs *store.ChainStore, _ dtypes.AfterGenesisSet) (dtypes.NetworkName, error) {
if !build.Devnet {
return "testnetnet", nil
}
ctx := helpers.LifecycleCtx(mctx, lc)
netName, err := stmgr.GetNetworkName(ctx, stmgr.NewStateManager(cs), cs.GetHeaviestTipSet().ParentState())
return netName, err
}
func NewSyncer(lc fx.Lifecycle, ds dtypes.MetadataDS, sm *stmgr.StateManager, exchange exchange.Client, h host.Host, beacon beacon.RandomBeacon, verifier ffiwrapper.Verifier) (*chain.Syncer, error) {
syncer, err := chain.NewSyncer(ds, sm, exchange, h.ConnManager(), h.ID(), beacon, verifier)
type SyncerParams struct {
fx.In
Lifecycle fx.Lifecycle
MetadataDS dtypes.MetadataDS
StateManager *stmgr.StateManager
ChainXchg exchange.Client
SyncMgrCtor chain.SyncManagerCtor
Host host.Host
Beacon beacon.Schedule
Verifier ffiwrapper.Verifier
}
func NewSyncer(params SyncerParams) (*chain.Syncer, error) {
var (
lc = params.Lifecycle
ds = params.MetadataDS
sm = params.StateManager
ex = params.ChainXchg
smCtor = params.SyncMgrCtor
h = params.Host
b = params.Beacon
v = params.Verifier
)
syncer, err := chain.NewSyncer(ds, sm, ex, smCtor, h.ConnManager(), h.ID(), b, v)
if err != nil {
return nil, err
}
node/modules/client.go
+17 -4
View File
@@ -12,8 +12,9 @@ import (
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"
"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/retrievalmarket"
"github.com/filecoin-project/go-fil-markets/retrievalmarket/discovery"
retrievalimpl "github.com/filecoin-project/go-fil-markets/retrievalmarket/impl"
rmnet "github.com/filecoin-project/go-fil-markets/retrievalmarket/network"
"github.com/filecoin-project/go-fil-markets/storagemarket"
@@ -26,7 +27,9 @@ import (
"github.com/ipfs/go-datastore/namespace"
"github.com/libp2p/go-libp2p-core/host"
"github.com/filecoin-project/lotus/journal"
"github.com/filecoin-project/lotus/lib/blockstore"
"github.com/filecoin-project/lotus/markets"
marketevents "github.com/filecoin-project/lotus/markets/loggers"
"github.com/filecoin-project/lotus/markets/retrievaladapter"
"github.com/filecoin-project/lotus/node/impl/full"
@@ -110,15 +113,20 @@ func NewClientDealFunds(ds dtypes.MetadataDS) (ClientDealFunds, error) {
return funds.NewDealFunds(ds, datastore.NewKey("/marketfunds/client"))
}
func StorageClient(lc fx.Lifecycle, h host.Host, ibs dtypes.ClientBlockstore, mds dtypes.ClientMultiDstore, r repo.LockedRepo, dataTransfer dtypes.ClientDataTransfer, discovery *discovery.Local, deals dtypes.ClientDatastore, scn storagemarket.StorageClientNode, dealFunds ClientDealFunds) (storagemarket.StorageClient, error) {
func StorageClient(lc fx.Lifecycle, h host.Host, ibs dtypes.ClientBlockstore, mds dtypes.ClientMultiDstore, r repo.LockedRepo, dataTransfer dtypes.ClientDataTransfer, discovery *discoveryimpl.Local, deals dtypes.ClientDatastore, scn storagemarket.StorageClientNode, dealFunds ClientDealFunds) (storagemarket.StorageClient, error) {
net := smnet.NewFromLibp2pHost(h)
c, err := storageimpl.NewClient(net, ibs, mds, dataTransfer, discovery, deals, scn, dealFunds, storageimpl.DealPollingInterval(time.Second))
if err != nil {
return nil, err
}
c.OnReady(marketevents.ReadyLogger("storage client"))
lc.Append(fx.Hook{
OnStart: func(ctx context.Context) error {
c.SubscribeToEvents(marketevents.StorageClientLogger)
evtType := journal.J.RegisterEventType("markets/storage/client", "state_change")
c.SubscribeToEvents(markets.StorageClientJournaler(evtType))
return c.Start(ctx)
},
OnStop: func(context.Context) error {
@@ -129,7 +137,7 @@ func StorageClient(lc fx.Lifecycle, h host.Host, ibs dtypes.ClientBlockstore, md
}
// RetrievalClient creates a new retrieval client attached to the client blockstore
func RetrievalClient(lc fx.Lifecycle, h host.Host, mds dtypes.ClientMultiDstore, dt dtypes.ClientDataTransfer, payAPI payapi.PaychAPI, resolver retrievalmarket.PeerResolver, ds dtypes.MetadataDS, chainAPI full.ChainAPI, stateAPI full.StateAPI) (retrievalmarket.RetrievalClient, error) {
func RetrievalClient(lc fx.Lifecycle, h host.Host, mds dtypes.ClientMultiDstore, dt dtypes.ClientDataTransfer, payAPI payapi.PaychAPI, resolver discovery.PeerResolver, ds dtypes.MetadataDS, chainAPI full.ChainAPI, stateAPI full.StateAPI) (retrievalmarket.RetrievalClient, error) {
adapter := retrievaladapter.NewRetrievalClientNode(payAPI, chainAPI, stateAPI)
network := rmnet.NewFromLibp2pHost(h)
sc := storedcounter.New(ds, datastore.NewKey("/retr"))
@@ -137,10 +145,15 @@ func RetrievalClient(lc fx.Lifecycle, h host.Host, mds dtypes.ClientMultiDstore,
if err != nil {
return nil, err
}
client.OnReady(marketevents.ReadyLogger("retrieval client"))
lc.Append(fx.Hook{
OnStart: func(ctx context.Context) error {
client.SubscribeToEvents(marketevents.RetrievalClientLogger)
return nil
evtType := journal.J.RegisterEventType("markets/retrieval/client", "state_change")
client.SubscribeToEvents(markets.RetrievalClientJournaler(evtType))
return client.Start(ctx)
},
})
return client, nil
node/modules/core.go
+11 -12
View File
@@ -6,10 +6,10 @@ import (
"errors"
"io"
"io/ioutil"
"path/filepath"
"github.com/gbrlsnchs/jwt/v3"
logging "github.com/ipfs/go-log/v2"
"github.com/libp2p/go-libp2p-core/peer"
"github.com/libp2p/go-libp2p-core/peerstore"
record "github.com/libp2p/go-libp2p-record"
"golang.org/x/xerrors"
@@ -19,7 +19,6 @@ import (
"github.com/filecoin-project/lotus/api/apistruct"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/journal"
"github.com/filecoin-project/lotus/lib/addrutil"
"github.com/filecoin-project/lotus/node/modules/dtypes"
"github.com/filecoin-project/lotus/node/repo"
@@ -93,16 +92,16 @@ func BuiltinBootstrap() (dtypes.BootstrapPeers, error) {
return build.BuiltinBootstrap()
}
func DrandBootstrap(d dtypes.DrandConfig) (dtypes.DrandBootstrap, error) {
func DrandBootstrap(ds dtypes.DrandSchedule) (dtypes.DrandBootstrap, error) {
// TODO: retry resolving, don't fail if at least one resolve succeeds
addrs, err := addrutil.ParseAddresses(context.TODO(), d.Relays)
if err != nil {
log.Errorf("reoslving drand relays addresses: %+v", err)
return nil, nil
res := []peer.AddrInfo{}
for _, d := range ds {
addrs, err := addrutil.ParseAddresses(context.TODO(), d.Config.Relays)
if err != nil {
log.Errorf("reoslving drand relays addresses: %+v", err)
return res, nil
}
res = append(res, addrs...)
}
return addrs, nil
}
func SetupJournal(lr repo.LockedRepo) error {
return journal.InitializeSystemJournal(filepath.Join(lr.Path(), "journal"))
return res, nil
}
node/modules/dtypes/beacon.go
+9
View File
@@ -1,5 +1,14 @@
package dtypes
import "github.com/filecoin-project/go-state-types/abi"
type DrandSchedule []DrandPoint
type DrandPoint struct {
Start abi.ChainEpoch
Config DrandConfig
}
type DrandConfig struct {
Servers []string
Relays []string
node/modules/lp2p/pubsub.go
+122 -114
View File
@@ -49,7 +49,7 @@ type GossipIn struct {
Db dtypes.DrandBootstrap
Cfg *config.Pubsub
Sk *dtypes.ScoreKeeper
Dr dtypes.DrandConfig
Dr dtypes.DrandSchedule
}
func getDrandTopic(chainInfoJSON string) (string, error) {
@@ -74,9 +74,126 @@ func GossipSub(in GossipIn) (service *pubsub.PubSub, err error) {
}
isBootstrapNode := in.Cfg.Bootstrapper
drandTopic, err := getDrandTopic(in.Dr.ChainInfoJSON)
if err != nil {
return nil, err
drandTopicParams := &pubsub.TopicScoreParams{
// expected 2 beaconsn/min
TopicWeight: 0.5, // 5x block topic; max cap is 62.5
// 1 tick per second, maxes at 1 after 1 hour
TimeInMeshWeight: 0.00027, // ~1/3600
TimeInMeshQuantum: time.Second,
TimeInMeshCap: 1,
// deliveries decay after 1 hour, cap at 25 beacons
FirstMessageDeliveriesWeight: 5, // max value is 125
FirstMessageDeliveriesDecay: pubsub.ScoreParameterDecay(time.Hour),
FirstMessageDeliveriesCap: 25, // the maximum expected in an hour is ~26, including the decay
// Mesh Delivery Failure is currently turned off for beacons
// This is on purpose as
// - the traffic is very low for meaningful distribution of incoming edges.
// - the reaction time needs to be very slow -- in the order of 10 min at least
// so we might as well let opportunistic grafting repair the mesh on its own
// pace.
// - the network is too small, so large asymmetries can be expected between mesh
// edges.
// We should revisit this once the network grows.
// invalid messages decay after 1 hour
InvalidMessageDeliveriesWeight: -1000,
InvalidMessageDeliveriesDecay: pubsub.ScoreParameterDecay(time.Hour),
}
topicParams := map[string]*pubsub.TopicScoreParams{
build.BlocksTopic(in.Nn): {
// expected 10 blocks/min
TopicWeight: 0.1, // max cap is 50, max mesh penalty is -10, single invalid message is -100
// 1 tick per second, maxes at 1 after 1 hour
TimeInMeshWeight: 0.00027, // ~1/3600
TimeInMeshQuantum: time.Second,
TimeInMeshCap: 1,
// deliveries decay after 1 hour, cap at 100 blocks
FirstMessageDeliveriesWeight: 5, // max value is 500
FirstMessageDeliveriesDecay: pubsub.ScoreParameterDecay(time.Hour),
FirstMessageDeliveriesCap: 100, // 100 blocks in an hour
// Mesh Delivery Failure is currently turned off for blocks
// This is on purpose as
// - the traffic is very low for meaningful distribution of incoming edges.
// - the reaction time needs to be very slow -- in the order of 10 min at least
// so we might as well let opportunistic grafting repair the mesh on its own
// pace.
// - the network is too small, so large asymmetries can be expected between mesh
// edges.
// We should revisit this once the network grows.
//
// // tracks deliveries in the last minute
// // penalty activates at 1 minute and expects ~0.4 blocks
// MeshMessageDeliveriesWeight: -576, // max penalty is -100
// MeshMessageDeliveriesDecay: pubsub.ScoreParameterDecay(time.Minute),
// MeshMessageDeliveriesCap: 10, // 10 blocks in a minute
// MeshMessageDeliveriesThreshold: 0.41666, // 10/12/2 blocks/min
// MeshMessageDeliveriesWindow: 10 * time.Millisecond,
// MeshMessageDeliveriesActivation: time.Minute,
//
// // decays after 15 min
// MeshFailurePenaltyWeight: -576,
// MeshFailurePenaltyDecay: pubsub.ScoreParameterDecay(15 * time.Minute),
// invalid messages decay after 1 hour
InvalidMessageDeliveriesWeight: -1000,
InvalidMessageDeliveriesDecay: pubsub.ScoreParameterDecay(time.Hour),
},
build.MessagesTopic(in.Nn): {
// expected > 1 tx/second
TopicWeight: 0.1, // max cap is 5, single invalid message is -100
// 1 tick per second, maxes at 1 hour
TimeInMeshWeight: 0.0002778, // ~1/3600
TimeInMeshQuantum: time.Second,
TimeInMeshCap: 1,
// deliveries decay after 10min, cap at 100 tx
FirstMessageDeliveriesWeight: 0.5, // max value is 50
FirstMessageDeliveriesDecay: pubsub.ScoreParameterDecay(10 * time.Minute),
FirstMessageDeliveriesCap: 100, // 100 messages in 10 minutes
// Mesh Delivery Failure is currently turned off for messages
// This is on purpose as the network is still too small, which results in
// asymmetries and potential unmeshing from negative scores.
// // tracks deliveries in the last minute
// // penalty activates at 1 min and expects 2.5 txs
// MeshMessageDeliveriesWeight: -16, // max penalty is -100
// MeshMessageDeliveriesDecay: pubsub.ScoreParameterDecay(time.Minute),
// MeshMessageDeliveriesCap: 100, // 100 txs in a minute
// MeshMessageDeliveriesThreshold: 2.5, // 60/12/2 txs/minute
// MeshMessageDeliveriesWindow: 10 * time.Millisecond,
// MeshMessageDeliveriesActivation: time.Minute,
// // decays after 5min
// MeshFailurePenaltyWeight: -16,
// MeshFailurePenaltyDecay: pubsub.ScoreParameterDecay(5 * time.Minute),
// invalid messages decay after 1 hour
InvalidMessageDeliveriesWeight: -1000,
InvalidMessageDeliveriesDecay: pubsub.ScoreParameterDecay(time.Hour),
},
}
pgTopicWeights := map[string]float64{
build.BlocksTopic(in.Nn): 10,
build.MessagesTopic(in.Nn): 1,
}
for _, d := range in.Dr {
topic, err := getDrandTopic(d.Config.ChainInfoJSON)
if err != nil {
return nil, err
}
topicParams[topic] = drandTopicParams
pgTopicWeights[topic] = 5
}
options := []pubsub.Option{
@@ -124,111 +241,7 @@ func GossipSub(in GossipIn) (service *pubsub.PubSub, err error) {
RetainScore: 6 * time.Hour,
// topic parameters
Topics: map[string]*pubsub.TopicScoreParams{
drandTopic: {
// expected 2 beaconsn/min
TopicWeight: 0.5, // 5x block topic; max cap is 62.5
// 1 tick per second, maxes at 1 after 1 hour
TimeInMeshWeight: 0.00027, // ~1/3600
TimeInMeshQuantum: time.Second,
TimeInMeshCap: 1,
// deliveries decay after 1 hour, cap at 25 beacons
FirstMessageDeliveriesWeight: 5, // max value is 125
FirstMessageDeliveriesDecay: pubsub.ScoreParameterDecay(time.Hour),
FirstMessageDeliveriesCap: 25, // the maximum expected in an hour is ~26, including the decay
// Mesh Delivery Failure is currently turned off for beacons
// This is on purpose as
// - the traffic is very low for meaningful distribution of incoming edges.
// - the reaction time needs to be very slow -- in the order of 10 min at least
// so we might as well let opportunistic grafting repair the mesh on its own
// pace.
// - the network is too small, so large asymmetries can be expected between mesh
// edges.
// We should revisit this once the network grows.
// invalid messages decay after 1 hour
InvalidMessageDeliveriesWeight: -1000,
InvalidMessageDeliveriesDecay: pubsub.ScoreParameterDecay(time.Hour),
},
build.BlocksTopic(in.Nn): {
// expected 10 blocks/min
TopicWeight: 0.1, // max cap is 50, max mesh penalty is -10, single invalid message is -100
// 1 tick per second, maxes at 1 after 1 hour
TimeInMeshWeight: 0.00027, // ~1/3600
TimeInMeshQuantum: time.Second,
TimeInMeshCap: 1,
// deliveries decay after 1 hour, cap at 100 blocks
FirstMessageDeliveriesWeight: 5, // max value is 500
FirstMessageDeliveriesDecay: pubsub.ScoreParameterDecay(time.Hour),
FirstMessageDeliveriesCap: 100, // 100 blocks in an hour
// Mesh Delivery Failure is currently turned off for blocks
// This is on purpose as
// - the traffic is very low for meaningful distribution of incoming edges.
// - the reaction time needs to be very slow -- in the order of 10 min at least
// so we might as well let opportunistic grafting repair the mesh on its own
// pace.
// - the network is too small, so large asymmetries can be expected between mesh
// edges.
// We should revisit this once the network grows.
//
// // tracks deliveries in the last minute
// // penalty activates at 1 minute and expects ~0.4 blocks
// MeshMessageDeliveriesWeight: -576, // max penalty is -100
// MeshMessageDeliveriesDecay: pubsub.ScoreParameterDecay(time.Minute),
// MeshMessageDeliveriesCap: 10, // 10 blocks in a minute
// MeshMessageDeliveriesThreshold: 0.41666, // 10/12/2 blocks/min
// MeshMessageDeliveriesWindow: 10 * time.Millisecond,
// MeshMessageDeliveriesActivation: time.Minute,
//
// // decays after 15 min
// MeshFailurePenaltyWeight: -576,
// MeshFailurePenaltyDecay: pubsub.ScoreParameterDecay(15 * time.Minute),
// invalid messages decay after 1 hour
InvalidMessageDeliveriesWeight: -1000,
InvalidMessageDeliveriesDecay: pubsub.ScoreParameterDecay(time.Hour),
},
build.MessagesTopic(in.Nn): {
// expected > 1 tx/second
TopicWeight: 0.1, // max cap is 5, single invalid message is -100
// 1 tick per second, maxes at 1 hour
TimeInMeshWeight: 0.0002778, // ~1/3600
TimeInMeshQuantum: time.Second,
TimeInMeshCap: 1,
// deliveries decay after 10min, cap at 100 tx
FirstMessageDeliveriesWeight: 0.5, // max value is 50
FirstMessageDeliveriesDecay: pubsub.ScoreParameterDecay(10 * time.Minute),
FirstMessageDeliveriesCap: 100, // 100 messages in 10 minutes
// Mesh Delivery Failure is currently turned off for messages
// This is on purpose as the network is still too small, which results in
// asymmetries and potential unmeshing from negative scores.
// // tracks deliveries in the last minute
// // penalty activates at 1 min and expects 2.5 txs
// MeshMessageDeliveriesWeight: -16, // max penalty is -100
// MeshMessageDeliveriesDecay: pubsub.ScoreParameterDecay(time.Minute),
// MeshMessageDeliveriesCap: 100, // 100 txs in a minute
// MeshMessageDeliveriesThreshold: 2.5, // 60/12/2 txs/minute
// MeshMessageDeliveriesWindow: 10 * time.Millisecond,
// MeshMessageDeliveriesActivation: time.Minute,
// // decays after 5min
// MeshFailurePenaltyWeight: -16,
// MeshFailurePenaltyDecay: pubsub.ScoreParameterDecay(5 * time.Minute),
// invalid messages decay after 1 hour
InvalidMessageDeliveriesWeight: -1000,
InvalidMessageDeliveriesDecay: pubsub.ScoreParameterDecay(time.Hour),
},
},
Topics: topicParams,
},
&pubsub.PeerScoreThresholds{
GossipThreshold: -500,
@@ -278,11 +291,6 @@ func GossipSub(in GossipIn) (service *pubsub.PubSub, err error) {
}
// validation queue RED
pgTopicWeights := map[string]float64{
drandTopic: 5,
build.BlocksTopic(in.Nn): 10,
build.MessagesTopic(in.Nn): 1,
}
var pgParams *pubsub.PeerGaterParams
if isBootstrapNode {
node/modules/services.go
+49 -12
View File
@@ -1,6 +1,8 @@
package modules
import (
"context"
"github.com/ipfs/go-datastore"
"github.com/ipfs/go-datastore/namespace"
eventbus "github.com/libp2p/go-eventbus"
@@ -11,8 +13,9 @@ import (
"go.uber.org/fx"
"golang.org/x/xerrors"
"github.com/filecoin-project/go-fil-markets/retrievalmarket"
"github.com/filecoin-project/go-fil-markets/retrievalmarket/discovery"
"github.com/filecoin-project/go-fil-markets/discovery"
discoveryimpl "github.com/filecoin-project/go-fil-markets/discovery/impl"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain"
"github.com/filecoin-project/lotus/chain/beacon"
@@ -22,10 +25,13 @@ import (
"github.com/filecoin-project/lotus/chain/stmgr"
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/sub"
"github.com/filecoin-project/lotus/journal"
"github.com/filecoin-project/lotus/lib/peermgr"
marketevents "github.com/filecoin-project/lotus/markets/loggers"
"github.com/filecoin-project/lotus/node/hello"
"github.com/filecoin-project/lotus/node/modules/dtypes"
"github.com/filecoin-project/lotus/node/modules/helpers"
"github.com/filecoin-project/lotus/node/repo"
)
func RunHello(mctx helpers.MetricsCtx, lc fx.Lifecycle, h host.Host, svc *hello.Service) error {
@@ -113,12 +119,22 @@ func HandleIncomingMessages(mctx helpers.MetricsCtx, lc fx.Lifecycle, ps *pubsub
go sub.HandleIncomingMessages(ctx, mpool, msgsub)
}
func NewLocalDiscovery(ds dtypes.MetadataDS) *discovery.Local {
return discovery.NewLocal(namespace.Wrap(ds, datastore.NewKey("/deals/local")))
func NewLocalDiscovery(lc fx.Lifecycle, ds dtypes.MetadataDS) (*discoveryimpl.Local, error) {
local, err := discoveryimpl.NewLocal(namespace.Wrap(ds, datastore.NewKey("/deals/local")))
if err != nil {
return nil, err
}
local.OnReady(marketevents.ReadyLogger("discovery"))
lc.Append(fx.Hook{
OnStart: func(ctx context.Context) error {
return local.Start(ctx)
},
})
return local, nil
}
func RetrievalResolver(l *discovery.Local) retrievalmarket.PeerResolver {
return discovery.Multi(l)
func RetrievalResolver(l *discoveryimpl.Local) discovery.PeerResolver {
return discoveryimpl.Multi(l)
}
type RandomBeaconParams struct {
@@ -126,19 +142,40 @@ type RandomBeaconParams struct {
PubSub *pubsub.PubSub `optional:"true"`
Cs *store.ChainStore
DrandConfig dtypes.DrandConfig
DrandConfig dtypes.DrandSchedule
}
func BuiltinDrandConfig() dtypes.DrandConfig {
return build.DrandConfig()
func BuiltinDrandConfig() dtypes.DrandSchedule {
return build.DrandConfigSchedule()
}
func RandomBeacon(p RandomBeaconParams, _ dtypes.AfterGenesisSet) (beacon.RandomBeacon, error) {
func RandomSchedule(p RandomBeaconParams, _ dtypes.AfterGenesisSet) (beacon.Schedule, error) {
gen, err := p.Cs.GetGenesis()
if err != nil {
return nil, err
}
//return beacon.NewMockBeacon(build.BlockDelaySecs * time.Second)
return drand.NewDrandBeacon(gen.Timestamp, build.BlockDelaySecs, p.PubSub, p.DrandConfig)
shd := beacon.Schedule{}
for _, dc := range p.DrandConfig {
bc, err := drand.NewDrandBeacon(gen.Timestamp, build.BlockDelaySecs, p.PubSub, dc.Config)
if err != nil {
return nil, xerrors.Errorf("creating drand beacon: %w", err)
}
shd = append(shd, beacon.BeaconPoint{Start: dc.Start, Beacon: bc})
}
return shd, nil
}
func OpenFilesystemJournal(lr repo.LockedRepo, lc fx.Lifecycle, disabled journal.DisabledEvents) (journal.Journal, error) {
jrnl, err := journal.OpenFSJournal(lr, disabled)
if err != nil {
return nil, err
}
lc.Append(fx.Hook{
OnStop: func(_ context.Context) error { return jrnl.Close() },
})
return jrnl, err
}
node/modules/storageminer.go
+67 -9
View File
@@ -29,10 +29,11 @@ import (
dtnet "github.com/filecoin-project/go-data-transfer/network"
dtgstransport "github.com/filecoin-project/go-data-transfer/transport/graphsync"
piecefilestore "github.com/filecoin-project/go-fil-markets/filestore"
"github.com/filecoin-project/go-fil-markets/piecestore"
piecestoreimpl "github.com/filecoin-project/go-fil-markets/piecestore/impl"
"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"
"github.com/filecoin-project/go-fil-markets/shared"
"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/funds"
@@ -49,6 +50,8 @@ import (
"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/journal"
"github.com/filecoin-project/lotus/markets"
lapi "github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/build"
@@ -107,6 +110,9 @@ func MinerID(ma dtypes.MinerAddress) (dtypes.MinerID, error) {
}
func StorageNetworkName(ctx helpers.MetricsCtx, a lapi.FullNode) (dtypes.NetworkName, error) {
if !build.Devnet {
return "testnetnet", nil
}
return a.StateNetworkName(ctx)
}
@@ -142,8 +148,34 @@ func SectorIDCounter(ds dtypes.MetadataDS) sealing.SectorIDCounter {
return &sidsc{sc}
}
func StorageMiner(fc config.MinerFeeConfig) func(mctx helpers.MetricsCtx, lc fx.Lifecycle, api lapi.FullNode, h host.Host, ds dtypes.MetadataDS, sealer sectorstorage.SectorManager, sc sealing.SectorIDCounter, verif ffiwrapper.Verifier, gsd dtypes.GetSealingConfigFunc) (*storage.Miner, error) {
return func(mctx helpers.MetricsCtx, lc fx.Lifecycle, api lapi.FullNode, h host.Host, ds dtypes.MetadataDS, sealer sectorstorage.SectorManager, sc sealing.SectorIDCounter, verif ffiwrapper.Verifier, gsd dtypes.GetSealingConfigFunc) (*storage.Miner, error) {
type StorageMinerParams struct {
fx.In
Lifecycle fx.Lifecycle
MetricsCtx helpers.MetricsCtx
API lapi.FullNode
Host host.Host
MetadataDS dtypes.MetadataDS
Sealer sectorstorage.SectorManager
SectorIDCounter sealing.SectorIDCounter
Verifier ffiwrapper.Verifier
GetSealingConfigFn dtypes.GetSealingConfigFunc
}
func StorageMiner(fc config.MinerFeeConfig) func(params StorageMinerParams) (*storage.Miner, error) {
return func(params StorageMinerParams) (*storage.Miner, error) {
var (
ds = params.MetadataDS
mctx = params.MetricsCtx
lc = params.Lifecycle
api = params.API
sealer = params.Sealer
h = params.Host
sc = params.SectorIDCounter
verif = params.Verifier
gsd = params.GetSealingConfigFn
)
maddr, err := minerAddrFromDS(ds)
if err != nil {
return nil, err
@@ -184,10 +216,16 @@ func StorageMiner(fc config.MinerFeeConfig) func(mctx helpers.MetricsCtx, lc fx.
}
func HandleRetrieval(host host.Host, lc fx.Lifecycle, m retrievalmarket.RetrievalProvider) {
m.OnReady(marketevents.ReadyLogger("retrieval provider"))
lc.Append(fx.Hook{
OnStart: func(context.Context) error {
OnStart: func(ctx context.Context) error {
m.SubscribeToEvents(marketevents.RetrievalProviderLogger)
return m.Start()
evtType := journal.J.RegisterEventType("markets/retrieval/provider", "state_change")
m.SubscribeToEvents(markets.RetrievalProviderJournaler(evtType))
return m.Start(ctx)
},
OnStop: func(context.Context) error {
return m.Stop()
@@ -197,10 +235,14 @@ func HandleRetrieval(host host.Host, lc fx.Lifecycle, m retrievalmarket.Retrieva
func HandleDeals(mctx helpers.MetricsCtx, lc fx.Lifecycle, host host.Host, h storagemarket.StorageProvider) {
ctx := helpers.LifecycleCtx(mctx, lc)
h.OnReady(marketevents.ReadyLogger("storage provider"))
lc.Append(fx.Hook{
OnStart: func(context.Context) error {
h.SubscribeToEvents(marketevents.StorageProviderLogger)
evtType := journal.J.RegisterEventType("markets/storage/provider", "state_change")
h.SubscribeToEvents(markets.StorageProviderJournaler(evtType))
return h.Start(ctx)
},
OnStop: func(context.Context) error {
@@ -235,8 +277,18 @@ func NewProviderDAGServiceDataTransfer(lc fx.Lifecycle, h host.Host, gs dtypes.S
// 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(namespace.Wrap(ds, datastore.NewKey("/storagemarket")))
func NewProviderPieceStore(lc fx.Lifecycle, ds dtypes.MetadataDS) (dtypes.ProviderPieceStore, error) {
ps, err := piecestoreimpl.NewPieceStore(namespace.Wrap(ds, datastore.NewKey("/storagemarket")))
if err != nil {
return nil, err
}
ps.OnReady(marketevents.ReadyLogger("piecestore"))
lc.Append(fx.Hook{
OnStart: func(ctx context.Context) error {
return ps.Start(ctx)
},
})
return ps, nil
}
func StagingMultiDatastore(lc fx.Lifecycle, r repo.LockedRepo) (dtypes.StagingMultiDstore, error) {
@@ -331,7 +383,13 @@ func NewStorageAsk(ctx helpers.MetricsCtx, fapi lapi.FullNode, ds dtypes.Metadat
return nil, err
}
storedAsk, err := storedask.NewStoredAsk(namespace.Wrap(ds, datastore.NewKey("/deals/provider")), datastore.NewKey("latest-ask"), spn, address.Address(minerAddress))
providerDs := namespace.Wrap(ds, datastore.NewKey("/deals/provider"))
// legacy this was mistake where this key was place -- so we move the legacy key if need be
err = shared.MoveKey(providerDs, "/latest-ask", "/storage-ask/latest")
if err != nil {
return nil, err
}
storedAsk, err := storedask.NewStoredAsk(namespace.Wrap(providerDs, datastore.NewKey("/storage-ask")), datastore.NewKey("latest"), spn, address.Address(minerAddress))
if err != nil {
return nil, err
}
node/modules/testing/beacon.go
+5 -2
View File
@@ -7,6 +7,9 @@ import (
"github.com/filecoin-project/lotus/chain/beacon"
)
func RandomBeacon() (beacon.RandomBeacon, error) {
return beacon.NewMockBeacon(time.Duration(build.BlockDelaySecs) * time.Second), nil
func RandomBeacon() (beacon.Schedule, error) {
return beacon.Schedule{
{Start: 0,
Beacon: beacon.NewMockBeacon(time.Duration(build.BlockDelaySecs) * time.Second),
}}, nil
}
node/node_test.go
+10 -10
View File
@@ -8,24 +8,19 @@ import (
builder "github.com/filecoin-project/lotus/node/test"
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/go-state-types/big"
"github.com/filecoin-project/lotus/lib/lotuslog"
saminer "github.com/filecoin-project/specs-actors/actors/builtin/miner"
"github.com/filecoin-project/specs-actors/actors/builtin/power"
"github.com/filecoin-project/specs-actors/actors/builtin/verifreg"
logging "github.com/ipfs/go-log/v2"
"github.com/filecoin-project/lotus/api/test"
"github.com/filecoin-project/lotus/chain/actors/policy"
)
func init() {
_ = logging.SetLogLevel("*", "INFO")
power.ConsensusMinerMinPower = big.NewInt(2048)
saminer.SupportedProofTypes = map[abi.RegisteredSealProof]struct{}{
abi.RegisteredSealProof_StackedDrg2KiBV1: {},
}
verifreg.MinVerifiedDealSize = big.NewInt(256)
policy.SetConsensusMinerMinPower(abi.NewStoragePower(2048))
policy.SetSupportedProofTypes(abi.RegisteredSealProof_StackedDrg2KiBV1)
policy.SetMinVerifiedDealSize(abi.NewStoragePower(256))
}
func TestAPI(t *testing.T) {
@@ -68,7 +63,12 @@ func TestAPIDealFlowReal(t *testing.T) {
logging.SetLogLevel("sub", "ERROR")
logging.SetLogLevel("storageminer", "ERROR")
saminer.PreCommitChallengeDelay = 5
// TODO: just set this globally?
oldDelay := policy.GetPreCommitChallengeDelay()
policy.SetPreCommitChallengeDelay(5)
t.Cleanup(func() {
policy.SetPreCommitChallengeDelay(oldDelay)
})
t.Run("basic", func(t *testing.T) {
test.TestDealFlow(t, builder.Builder, time.Second, false, false)
node/repo/fsrepo.go
+21 -15
View File
@@ -65,7 +65,8 @@ var ErrRepoExists = xerrors.New("repo exists")
// FsRepo is struct for repo, use NewFS to create
type FsRepo struct {
path string
path string
configPath string
}
var _ Repo = &FsRepo{}
@@ -78,10 +79,15 @@ func NewFS(path string) (*FsRepo, error) {
}
return &FsRepo{
path: path,
path: path,
configPath: filepath.Join(path, fsConfig),
}, nil
}
func (fsr *FsRepo) SetConfigPath(cfgPath string) {
fsr.configPath = cfgPath
}
func (fsr *FsRepo) Exists() (bool, error) {
_, err := os.Stat(filepath.Join(fsr.path, fsDatastore))
notexist := os.IsNotExist(err)
@@ -101,7 +107,7 @@ func (fsr *FsRepo) Init(t RepoType) error {
}
log.Infof("Initializing repo at '%s'", fsr.path)
err = os.Mkdir(fsr.path, 0755) //nolint: gosec
err = os.MkdirAll(fsr.path, 0755) //nolint: gosec
if err != nil && !os.IsExist(err) {
return err
}
@@ -115,9 +121,7 @@ func (fsr *FsRepo) Init(t RepoType) error {
}
func (fsr *FsRepo) initConfig(t RepoType) error {
cfgP := filepath.Join(fsr.path, fsConfig)
_, err := os.Stat(cfgP)
_, err := os.Stat(fsr.configPath)
if err == nil {
// exists
return nil
@@ -125,7 +129,7 @@ func (fsr *FsRepo) initConfig(t RepoType) error {
return err
}
c, err := os.Create(cfgP)
c, err := os.Create(fsr.configPath)
if err != nil {
return err
}
@@ -215,16 +219,18 @@ func (fsr *FsRepo) Lock(repoType RepoType) (LockedRepo, error) {
return nil, xerrors.Errorf("could not lock the repo: %w", err)
}
return &fsLockedRepo{
path: fsr.path,
repoType: repoType,
closer: closer,
path: fsr.path,
configPath: fsr.configPath,
repoType: repoType,
closer: closer,
}, nil
}
type fsLockedRepo struct {
path string
repoType RepoType
closer io.Closer
path string
configPath string
repoType RepoType
closer io.Closer
ds map[string]datastore.Batching
dsErr error
@@ -277,7 +283,7 @@ func (fsr *fsLockedRepo) Config() (interface{}, error) {
}
func (fsr *fsLockedRepo) loadConfigFromDisk() (interface{}, error) {
return config.FromFile(fsr.join(fsConfig), defConfForType(fsr.repoType))
return config.FromFile(fsr.configPath, defConfForType(fsr.repoType))
}
func (fsr *fsLockedRepo) SetConfig(c func(interface{})) error {
@@ -306,7 +312,7 @@ func (fsr *fsLockedRepo) SetConfig(c func(interface{})) error {
}
// write buffer of TOML bytes to config file
err = ioutil.WriteFile(fsr.join(fsConfig), buf.Bytes(), 0644)
err = ioutil.WriteFile(fsr.configPath, buf.Bytes(), 0644)
if err != nil {
return err
}
node/test/builder.go
+2 -2
View File
@@ -37,7 +37,7 @@ import (
"github.com/filecoin-project/lotus/node/repo"
"github.com/filecoin-project/lotus/storage/mockstorage"
"github.com/filecoin-project/specs-actors/actors/builtin"
"github.com/filecoin-project/specs-actors/actors/builtin/miner"
miner0 "github.com/filecoin-project/specs-actors/actors/builtin/miner"
"github.com/ipfs/go-datastore"
"github.com/libp2p/go-libp2p-core/crypto"
"github.com/libp2p/go-libp2p-core/peer"
@@ -83,7 +83,7 @@ func CreateTestStorageNode(ctx context.Context, t *testing.T, waddr address.Addr
peerid, err := peer.IDFromPrivateKey(pk)
require.NoError(t, err)
enc, err := actors.SerializeParams(&miner.ChangePeerIDParams{NewID: abi.PeerID(peerid)})
enc, err := actors.SerializeParams(&miner0.ChangePeerIDParams{NewID: abi.PeerID(peerid)})
require.NoError(t, err)
msg := &types.Message{