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

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

Browse Source
This commit is contained in:
Peter Rabbitson
2021-05-06 15:57:10 +02:00
parent b8dff22a40 c074031fa1
commit e648b72d8d
262 changed files with 23906 additions and 2118 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node/builder.go
+8 -5
View File
@@ -18,7 +18,7 @@ import (
"github.com/filecoin-project/lotus/node/hello"
"github.com/filecoin-project/lotus/system"
logging "github.com/ipfs/go-log"
logging "github.com/ipfs/go-log/v2"
ci "github.com/libp2p/go-libp2p-core/crypto"
"github.com/libp2p/go-libp2p-core/host"
"github.com/libp2p/go-libp2p-core/peer"
@@ -612,18 +612,21 @@ func Repo(r repo.Repo) Option {
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.BasicChainBlockstore), modules.ChainSplitBlockstore),
Override(new(dtypes.BasicStateBlockstore), 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.BasicChainBlockstore), modules.ChainFlatBlockstore),
Override(new(dtypes.BasicStateBlockstore), modules.StateFlatBlockstore),
Override(new(dtypes.BaseBlockstore), From(new(dtypes.UniversalBlockstore))),
Override(new(dtypes.ExposedBlockstore), From(new(dtypes.UniversalBlockstore))),
),
Override(new(dtypes.ChainBlockstore), From(new(dtypes.BasicChainBlockstore))),
Override(new(dtypes.StateBlockstore), From(new(dtypes.BasicStateBlockstore))),
If(os.Getenv("LOTUS_ENABLE_CHAINSTORE_FALLBACK") == "1",
Override(new(dtypes.ChainBlockstore), modules.FallbackChainBlockstore),
Override(new(dtypes.StateBlockstore), modules.FallbackStateBlockstore),
node/config/def.go
+1 -1
View File
@@ -205,7 +205,7 @@ func defCommon() Common {
}
var DefaultDefaultMaxFee = types.MustParseFIL("0.007")
var DefaultDefaultMaxFee = types.MustParseFIL("0.07")
var DefaultSimultaneousTransfers = uint64(20)
// DefaultFullNode returns the default config
node/impl/client/client.go
+5
View File
@@ -702,6 +702,11 @@ func (a *API) clientRetrieve(ctx context.Context, order api.RetrievalOrder, ref
}
}
if order.Total.Int == nil {
finish(xerrors.Errorf("cannot make retrieval deal for null total"))
return
}
if order.Size == 0 {
finish(xerrors.Errorf("cannot make retrieval deal for zero bytes"))
return
node/impl/full/chain.go
+3 -1
View File
@@ -18,7 +18,7 @@ import (
offline "github.com/ipfs/go-ipfs-exchange-offline"
cbor "github.com/ipfs/go-ipld-cbor"
ipld "github.com/ipfs/go-ipld-format"
logging "github.com/ipfs/go-log"
logging "github.com/ipfs/go-log/v2"
"github.com/ipfs/go-merkledag"
"github.com/ipfs/go-path"
"github.com/ipfs/go-path/resolver"
@@ -51,6 +51,8 @@ type ChainModuleAPI interface {
ChainReadObj(context.Context, cid.Cid) ([]byte, error)
}
var _ ChainModuleAPI = *new(api.FullNode)
// ChainModule provides a default implementation of ChainModuleAPI.
// It can be swapped out with another implementation through Dependency
// Injection (for example with a thin RPC client).
node/impl/full/gas.go
+4 -2
View File
@@ -31,6 +31,8 @@ type GasModuleAPI interface {
GasEstimateMessageGas(ctx context.Context, msg *types.Message, spec *api.MessageSendSpec, tsk types.TipSetKey) (*types.Message, error)
}
var _ GasModuleAPI = *new(api.FullNode)
// GasModule provides a default implementation of GasModuleAPI.
// It can be swapped out with another implementation through Dependency
// Injection (for example with a thin RPC client).
@@ -322,7 +324,7 @@ func gasEstimateGasLimit(
func (m *GasModule) GasEstimateMessageGas(ctx context.Context, msg *types.Message, spec *api.MessageSendSpec, _ types.TipSetKey) (*types.Message, error) {
if msg.GasLimit == 0 {
gasLimit, err := m.GasEstimateGasLimit(ctx, msg, types.TipSetKey{})
gasLimit, err := m.GasEstimateGasLimit(ctx, msg, types.EmptyTSK)
if err != nil {
return nil, xerrors.Errorf("estimating gas used: %w", err)
}
@@ -330,7 +332,7 @@ func (m *GasModule) GasEstimateMessageGas(ctx context.Context, msg *types.Messag
}
if msg.GasPremium == types.EmptyInt || types.BigCmp(msg.GasPremium, types.NewInt(0)) == 0 {
gasPremium, err := m.GasEstimateGasPremium(ctx, 10, msg.From, msg.GasLimit, types.TipSetKey{})
gasPremium, err := m.GasEstimateGasPremium(ctx, 10, msg.From, msg.GasLimit, types.EmptyTSK)
if err != nil {
return nil, xerrors.Errorf("estimating gas price: %w", err)
}
node/impl/full/mpool.go
+3 -1
View File
@@ -20,6 +20,8 @@ type MpoolModuleAPI interface {
MpoolPush(ctx context.Context, smsg *types.SignedMessage) (cid.Cid, error)
}
var _ MpoolModuleAPI = *new(api.FullNode)
// MpoolModule provides a default implementation of MpoolModuleAPI.
// It can be swapped out with another implementation through Dependency
// Injection (for example with a thin RPC client).
@@ -224,7 +226,7 @@ func (a *MpoolAPI) MpoolBatchPushMessage(ctx context.Context, msgs []*types.Mess
}
func (a *MpoolAPI) MpoolGetNonce(ctx context.Context, addr address.Address) (uint64, error) {
return a.Mpool.GetNonce(addr)
return a.Mpool.GetNonce(ctx, addr, types.EmptyTSK)
}
func (a *MpoolAPI) MpoolSub(ctx context.Context) (<-chan api.MpoolUpdate, error) {
node/impl/full/state.go
+39 -30
View File
@@ -44,7 +44,6 @@ type StateModuleAPI interface {
StateAccountKey(ctx context.Context, addr address.Address, tsk types.TipSetKey) (address.Address, error)
StateDealProviderCollateralBounds(ctx context.Context, size abi.PaddedPieceSize, verified bool, tsk types.TipSetKey) (api.DealCollateralBounds, error)
StateGetActor(ctx context.Context, actor address.Address, tsk types.TipSetKey) (*types.Actor, error)
StateGetReceipt(context.Context, cid.Cid, types.TipSetKey) (*types.MessageReceipt, error)
StateListMiners(ctx context.Context, tsk types.TipSetKey) ([]address.Address, error)
StateLookupID(ctx context.Context, addr address.Address, tsk types.TipSetKey) (address.Address, error)
StateMarketBalance(ctx context.Context, addr address.Address, tsk types.TipSetKey) (api.MarketBalance, error)
@@ -53,12 +52,14 @@ type StateModuleAPI interface {
StateMinerProvingDeadline(ctx context.Context, addr address.Address, tsk types.TipSetKey) (*dline.Info, error)
StateMinerPower(context.Context, address.Address, types.TipSetKey) (*api.MinerPower, error)
StateNetworkVersion(ctx context.Context, key types.TipSetKey) (network.Version, error)
StateSearchMsg(ctx context.Context, msg cid.Cid) (*api.MsgLookup, error)
StateSectorGetInfo(ctx context.Context, maddr address.Address, n abi.SectorNumber, tsk types.TipSetKey) (*miner.SectorOnChainInfo, error)
StateVerifiedClientStatus(ctx context.Context, addr address.Address, tsk types.TipSetKey) (*abi.StoragePower, error)
StateWaitMsg(ctx context.Context, msg cid.Cid, confidence uint64) (*api.MsgLookup, error)
StateSearchMsg(ctx context.Context, from types.TipSetKey, msg cid.Cid, limit abi.ChainEpoch, allowReplaced bool) (*api.MsgLookup, error)
StateWaitMsg(ctx context.Context, cid cid.Cid, confidence uint64, limit abi.ChainEpoch, allowReplaced bool) (*api.MsgLookup, error)
}
var _ StateModuleAPI = *new(api.FullNode)
// StateModule provides a default implementation of StateModuleAPI.
// It can be swapped out with another implementation through Dependency
// Injection (for example with a thin RPC client).
@@ -378,7 +379,7 @@ func (a *StateAPI) StateReplay(ctx context.Context, tsk types.TipSetKey, mc cid.
var ts *types.TipSet
var err error
if tsk == types.EmptyTSK {
mlkp, err := a.StateSearchMsg(ctx, mc)
mlkp, err := a.StateSearchMsg(ctx, types.EmptyTSK, mc, stmgr.LookbackNoLimit, true)
if err != nil {
return nil, xerrors.Errorf("searching for msg %s: %w", mc, err)
}
@@ -520,28 +521,22 @@ func (a *StateAPI) MinerCreateBlock(ctx context.Context, bt *api.BlockTemplate)
return &out, nil
}
func (m *StateModule) StateWaitMsg(ctx context.Context, msg cid.Cid, confidence uint64) (*api.MsgLookup, error) {
return stateWaitMsgLimited(ctx, m.StateManager, m.Chain, msg, confidence, stmgr.LookbackNoLimit)
}
func (a *StateAPI) StateWaitMsgLimited(ctx context.Context, msg cid.Cid, confidence uint64, lookbackLimit abi.ChainEpoch) (*api.MsgLookup, error) {
return stateWaitMsgLimited(ctx, a.StateManager, a.Chain, msg, confidence, lookbackLimit)
}
func stateWaitMsgLimited(ctx context.Context, smgr *stmgr.StateManager, cstore *store.ChainStore, msg cid.Cid, confidence uint64, lookbackLimit abi.ChainEpoch) (*api.MsgLookup, error) {
ts, recpt, found, err := smgr.WaitForMessage(ctx, msg, confidence, lookbackLimit)
func (m *StateModule) StateWaitMsg(ctx context.Context, msg cid.Cid, confidence uint64, lookbackLimit abi.ChainEpoch, allowReplaced bool) (*api.MsgLookup, error) {
ts, recpt, found, err := m.StateManager.WaitForMessage(ctx, msg, confidence, lookbackLimit, allowReplaced)
if err != nil {
return nil, err
}
var returndec interface{}
if recpt.ExitCode == 0 && len(recpt.Return) > 0 {
cmsg, err := cstore.GetCMessage(msg)
cmsg, err := m.Chain.GetCMessage(msg)
if err != nil {
return nil, xerrors.Errorf("failed to load message after successful receipt search: %w", err)
}
vmsg := cmsg.VMMessage()
t, err := stmgr.GetReturnType(ctx, smgr, vmsg.To, vmsg.Method, ts)
t, err := stmgr.GetReturnType(ctx, m.StateManager, vmsg.To, vmsg.Method, ts)
if err != nil {
return nil, xerrors.Errorf("failed to get return type: %w", err)
}
@@ -562,14 +557,13 @@ func stateWaitMsgLimited(ctx context.Context, smgr *stmgr.StateManager, cstore *
}, nil
}
func (m *StateModule) StateSearchMsg(ctx context.Context, msg cid.Cid) (*api.MsgLookup, error) {
return stateSearchMsgLimited(ctx, m.StateManager, msg, stmgr.LookbackNoLimit)
}
func (a *StateAPI) StateSearchMsgLimited(ctx context.Context, msg cid.Cid, lookbackLimit abi.ChainEpoch) (*api.MsgLookup, error) {
return stateSearchMsgLimited(ctx, a.StateManager, msg, lookbackLimit)
}
func stateSearchMsgLimited(ctx context.Context, smgr *stmgr.StateManager, msg cid.Cid, lookbackLimit abi.ChainEpoch) (*api.MsgLookup, error) {
ts, recpt, found, err := smgr.SearchForMessage(ctx, msg, lookbackLimit)
func (m *StateModule) StateSearchMsg(ctx context.Context, tsk types.TipSetKey, msg cid.Cid, lookbackLimit abi.ChainEpoch, allowReplaced bool) (*api.MsgLookup, error) {
fromTs, err := m.Chain.GetTipSetFromKey(tsk)
if err != nil {
return nil, xerrors.Errorf("loading tipset %s: %w", tsk, err)
}
ts, recpt, found, err := m.StateManager.SearchForMessage(ctx, fromTs, msg, lookbackLimit, allowReplaced)
if err != nil {
return nil, err
}
@@ -585,14 +579,6 @@ func stateSearchMsgLimited(ctx context.Context, smgr *stmgr.StateManager, msg ci
return nil, nil
}
func (m *StateModule) StateGetReceipt(ctx context.Context, msg cid.Cid, tsk types.TipSetKey) (*types.MessageReceipt, error) {
ts, err := m.Chain.GetTipSetFromKey(tsk)
if err != nil {
return nil, xerrors.Errorf("loading tipset %s: %w", tsk, err)
}
return m.StateManager.GetReceipt(ctx, msg, ts)
}
func (m *StateModule) StateListMiners(ctx context.Context, tsk types.TipSetKey) ([]address.Address, error) {
ts, err := m.Chain.GetTipSetFromKey(tsk)
if err != nil {
@@ -822,8 +808,31 @@ func (a *StateAPI) StateListMessages(ctx context.Context, match *api.MessageMatc
if match.To == address.Undef && match.From == address.Undef {
return nil, xerrors.Errorf("must specify at least To or From in message filter")
} else if match.To != address.Undef {
_, err := a.StateLookupID(ctx, match.To, tsk)
// if the recipient doesn't exist at the start point, we're not gonna find any matches
if xerrors.Is(err, types.ErrActorNotFound) {
return nil, nil
}
if err != nil {
return nil, xerrors.Errorf("looking up match.To: %w", err)
}
} else if match.From != address.Undef {
_, err := a.StateLookupID(ctx, match.From, tsk)
// if the sender doesn't exist at the start point, we're not gonna find any matches
if xerrors.Is(err, types.ErrActorNotFound) {
return nil, nil
}
if err != nil {
return nil, xerrors.Errorf("looking up match.From: %w", err)
}
}
// TODO: This should probably match on both ID and robust address, no?
matchFunc := func(msg *types.Message) bool {
if match.From != address.Undef && match.From != msg.From {
return false
node/impl/full/sync.go
+1 -1
View File
@@ -104,7 +104,7 @@ func (a *SyncAPI) SyncIncomingBlocks(ctx context.Context) (<-chan *types.BlockHe
func (a *SyncAPI) SyncCheckpoint(ctx context.Context, tsk types.TipSetKey) error {
log.Warnf("Marking tipset %s as checkpoint", tsk)
return a.Syncer.SetCheckpoint(tsk)
return a.Syncer.SyncCheckpoint(ctx, tsk)
}
func (a *SyncAPI) SyncMarkBad(ctx context.Context, bcid cid.Cid) error {
node/impl/remoteworker.go
+1 -1
View File
@@ -33,7 +33,7 @@ func connectRemoteWorker(ctx context.Context, fa api.Common, url string) (*remot
headers := http.Header{}
headers.Add("Authorization", "Bearer "+string(token))
wapi, closer, err := client.NewWorkerRPC(context.TODO(), url, headers)
wapi, closer, err := client.NewWorkerRPCV0(context.TODO(), url, headers)
if err != nil {
return nil, xerrors.Errorf("creating jsonrpc client: %w", err)
}
node/impl/storminer.go
+1 -2
View File
@@ -34,7 +34,6 @@ import (
sealing "github.com/filecoin-project/lotus/extern/storage-sealing"
"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"
@@ -90,7 +89,7 @@ type StorageMinerAPI struct {
}
func (sm *StorageMinerAPI) ServeRemote(w http.ResponseWriter, r *http.Request) {
if !auth.HasPerm(r.Context(), nil, apistruct.PermAdmin) {
if !auth.HasPerm(r.Context(), nil, api.PermAdmin) {
w.WriteHeader(401)
_ = json.NewEncoder(w).Encode(struct{ Error string }{"unauthorized: missing write permission"})
return
node/modules/blockstore.go
+4 -4
View File
@@ -94,11 +94,11 @@ func SplitBlockstore(cfg *config.Chainstore) func(lc fx.Lifecycle, r repo.Locked
}
}
func StateFlatBlockstore(_ fx.Lifecycle, _ helpers.MetricsCtx, bs dtypes.UniversalBlockstore) (dtypes.StateBlockstore, error) {
func StateFlatBlockstore(_ fx.Lifecycle, _ helpers.MetricsCtx, bs dtypes.UniversalBlockstore) (dtypes.BasicStateBlockstore, error) {
return bs, nil
}
func StateSplitBlockstore(_ fx.Lifecycle, _ helpers.MetricsCtx, bs dtypes.SplitBlockstore) (dtypes.StateBlockstore, error) {
func StateSplitBlockstore(_ fx.Lifecycle, _ helpers.MetricsCtx, bs dtypes.SplitBlockstore) (dtypes.BasicStateBlockstore, error) {
return bs, nil
}
@@ -110,11 +110,11 @@ func ChainSplitBlockstore(_ fx.Lifecycle, _ helpers.MetricsCtx, bs dtypes.SplitB
return bs, nil
}
func FallbackChainBlockstore(cbs dtypes.ChainBlockstore) dtypes.ChainBlockstore {
func FallbackChainBlockstore(cbs dtypes.BasicChainBlockstore) dtypes.ChainBlockstore {
return &blockstore.FallbackStore{Blockstore: cbs}
}
func FallbackStateBlockstore(sbs dtypes.StateBlockstore) dtypes.StateBlockstore {
func FallbackStateBlockstore(sbs dtypes.BasicStateBlockstore) dtypes.StateBlockstore {
return &blockstore.FallbackStore{Blockstore: sbs}
}
node/modules/client.go
+1 -3
View File
@@ -25,7 +25,6 @@ import (
smnet "github.com/filecoin-project/go-fil-markets/storagemarket/network"
"github.com/filecoin-project/go-multistore"
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/go-storedcounter"
"github.com/ipfs/go-datastore"
"github.com/ipfs/go-datastore/namespace"
"github.com/libp2p/go-libp2p-core/host"
@@ -210,8 +209,7 @@ func StorageClient(lc fx.Lifecycle, h host.Host, ibs dtypes.ClientBlockstore, md
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, j journal.Journal) (retrievalmarket.RetrievalClient, error) {
adapter := retrievaladapter.NewRetrievalClientNode(payAPI, chainAPI, stateAPI)
network := rmnet.NewFromLibp2pHost(h)
sc := storedcounter.New(ds, datastore.NewKey("/retr"))
client, err := retrievalimpl.NewClient(network, mds, dt, adapter, resolver, namespace.Wrap(ds, datastore.NewKey("/retrievals/client")), sc)
client, err := retrievalimpl.NewClient(network, mds, dt, adapter, resolver, namespace.Wrap(ds, datastore.NewKey("/retrievals/client")))
if err != nil {
return nil, err
}
node/modules/core.go
+2 -2
View File
@@ -22,7 +22,7 @@ import (
"github.com/filecoin-project/go-jsonrpc/auth"
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/lotus/api/apistruct"
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/lib/addrutil"
@@ -163,7 +163,7 @@ func APISecret(keystore types.KeyStore, lr repo.LockedRepo) (*dtypes.APIAlg, err
// TODO: make this configurable
p := JwtPayload{
Allow: apistruct.AllPermissions,
Allow: api.AllPermissions,
}
cliToken, err := jwt.Sign(&p, jwt.NewHS256(key.PrivateKey))
node/modules/dtypes/storage.go
+8
View File
@@ -36,12 +36,20 @@ type (
// BaseBlockstore is something, coz DI
BaseBlockstore blockstore.Blockstore
// BasicChainBlockstore is like ChainBlockstore, but without the optional
// network fallback support
BasicChainBlockstore 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
// patterns.
ChainBlockstore blockstore.Blockstore
// BasicStateBlockstore is like StateBlockstore, but without the optional
// network fallback support
BasicStateBlockstore blockstore.Blockstore
// StateBlockstore is a blockstore to store state data (state tree). It is
// physically backed by the BareMonolithBlockstore, but it has a cache on
// top that is specially tuned for state data access patterns.
node/modules/mpoolnonceapi.go
+60 -23
View File
@@ -2,6 +2,7 @@ package modules
import (
"context"
"strings"
"go.uber.org/fx"
"golang.org/x/xerrors"
@@ -19,41 +20,77 @@ import (
type MpoolNonceAPI struct {
fx.In
StateAPI full.StateAPI
ChainModule full.ChainModuleAPI
StateModule full.StateModuleAPI
}
// GetNonce gets the nonce from current chain head.
func (a *MpoolNonceAPI) GetNonce(addr address.Address) (uint64, error) {
ts := a.StateAPI.Chain.GetHeaviestTipSet()
func (a *MpoolNonceAPI) GetNonce(ctx context.Context, addr address.Address, tsk types.TipSetKey) (uint64, error) {
var err error
var ts *types.TipSet
if tsk == types.EmptyTSK {
// we need consistent tsk
ts, err = a.ChainModule.ChainHead(ctx)
if err != nil {
return 0, xerrors.Errorf("getting head: %w", err)
}
tsk = ts.Key()
} else {
ts, err = a.ChainModule.ChainGetTipSet(ctx, tsk)
if err != nil {
return 0, xerrors.Errorf("getting tipset: %w", err)
}
}
// make sure we have a key address so we can compare with messages
keyAddr, err := a.StateAPI.StateManager.ResolveToKeyAddress(context.TODO(), addr, ts)
if err != nil {
return 0, err
keyAddr := addr
if addr.Protocol() == address.ID {
// make sure we have a key address so we can compare with messages
keyAddr, err = a.StateModule.StateAccountKey(ctx, addr, tsk)
if err != nil {
return 0, xerrors.Errorf("getting account key: %w", err)
}
} else {
addr, err = a.StateModule.StateLookupID(ctx, addr, types.EmptyTSK)
if err != nil {
log.Infof("failed to look up id addr for %s: %w", addr, err)
addr = address.Undef
}
}
// Load the last nonce from the state, if it exists.
highestNonce := uint64(0)
if baseActor, err := a.StateAPI.StateManager.LoadActorRaw(context.TODO(), addr, ts.ParentState()); err != nil {
if !xerrors.Is(err, types.ErrActorNotFound) {
return 0, err
act, err := a.StateModule.StateGetActor(ctx, keyAddr, ts.Key())
if err != nil {
if strings.Contains(err.Error(), types.ErrActorNotFound.Error()) {
return 0, types.ErrActorNotFound
}
return 0, xerrors.Errorf("getting actor: %w", err)
}
highestNonce = act.Nonce
apply := func(msg *types.Message) {
if msg.From != addr && msg.From != keyAddr {
return
}
if msg.Nonce == highestNonce {
highestNonce = msg.Nonce + 1
}
} else {
highestNonce = baseActor.Nonce
}
// Otherwise, find the highest nonce in the tipset.
msgs, err := a.StateAPI.Chain.MessagesForTipset(ts)
if err != nil {
return 0, err
}
for _, msg := range msgs {
vmmsg := msg.VMMessage()
if vmmsg.From != keyAddr {
continue
for _, b := range ts.Blocks() {
msgs, err := a.ChainModule.ChainGetBlockMessages(ctx, b.Cid())
if err != nil {
return 0, xerrors.Errorf("getting block messages: %w", err)
}
if vmmsg.Nonce >= highestNonce {
highestNonce = vmmsg.Nonce + 1
if keyAddr.Protocol() == address.BLS {
for _, m := range msgs.BlsMessages {
apply(m)
}
} else {
for _, sm := range msgs.SecpkMessages {
apply(&sm.Message)
}
}
}
return highestNonce, nil
node/modules/paych.go
+2
View File
@@ -32,6 +32,8 @@ type PaychAPI struct {
full.StateAPI
}
var _ paychmgr.PaychAPI = &PaychAPI{}
// HandlePaychManager is called by dependency injection to set up hooks
func HandlePaychManager(lc fx.Lifecycle, pm *paychmgr.Manager) {
lc.Append(fx.Hook{
node/modules/storageminer.go
+9 -8
View File
@@ -55,7 +55,8 @@ import (
sealing "github.com/filecoin-project/lotus/extern/storage-sealing"
"github.com/filecoin-project/lotus/extern/storage-sealing/sealiface"
lapi "github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/api/v0api"
"github.com/filecoin-project/lotus/api/v1api"
"github.com/filecoin-project/lotus/blockstore"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain/actors/builtin"
@@ -116,14 +117,14 @@ func MinerID(ma dtypes.MinerAddress) (dtypes.MinerID, error) {
return dtypes.MinerID(id), err
}
func StorageNetworkName(ctx helpers.MetricsCtx, a lapi.FullNode) (dtypes.NetworkName, error) {
func StorageNetworkName(ctx helpers.MetricsCtx, a v1api.FullNode) (dtypes.NetworkName, error) {
if !build.Devnet {
return "testnetnet", nil
}
return a.StateNetworkName(ctx)
}
func SealProofType(maddr dtypes.MinerAddress, fnapi lapi.FullNode) (abi.RegisteredSealProof, error) {
func SealProofType(maddr dtypes.MinerAddress, fnapi v1api.FullNode) (abi.RegisteredSealProof, error) {
mi, err := fnapi.StateMinerInfo(context.TODO(), address.Address(maddr), types.EmptyTSK)
if err != nil {
return 0, err
@@ -196,7 +197,7 @@ type StorageMinerParams struct {
Lifecycle fx.Lifecycle
MetricsCtx helpers.MetricsCtx
API lapi.FullNode
API v1api.FullNode
Host host.Host
MetadataDS dtypes.MetadataDS
Sealer sectorstorage.SectorManager
@@ -437,7 +438,7 @@ func StagingGraphsync(mctx helpers.MetricsCtx, lc fx.Lifecycle, ibs dtypes.Stagi
return gs
}
func SetupBlockProducer(lc fx.Lifecycle, ds dtypes.MetadataDS, api lapi.FullNode, epp gen.WinningPoStProver, sf *slashfilter.SlashFilter, j journal.Journal) (*lotusminer.Miner, error) {
func SetupBlockProducer(lc fx.Lifecycle, ds dtypes.MetadataDS, api v1api.FullNode, epp gen.WinningPoStProver, sf *slashfilter.SlashFilter, j journal.Journal) (*lotusminer.Miner, error) {
minerAddr, err := minerAddrFromDS(ds)
if err != nil {
return nil, err
@@ -460,7 +461,7 @@ func SetupBlockProducer(lc fx.Lifecycle, ds dtypes.MetadataDS, api lapi.FullNode
return m, nil
}
func NewStorageAsk(ctx helpers.MetricsCtx, fapi lapi.FullNode, ds dtypes.MetadataDS, minerAddress dtypes.MinerAddress, spn storagemarket.StorageProviderNode) (*storedask.StoredAsk, error) {
func NewStorageAsk(ctx helpers.MetricsCtx, fapi v1api.FullNode, ds dtypes.MetadataDS, minerAddress dtypes.MinerAddress, spn storagemarket.StorageProviderNode) (*storedask.StoredAsk, error) {
mi, err := fapi.StateMinerInfo(ctx, address.Address(minerAddress), types.EmptyTSK)
if err != nil {
@@ -635,7 +636,7 @@ func RetrievalDealFilter(userFilter dtypes.RetrievalDealFilter) func(onlineOk dt
func RetrievalProvider(h host.Host,
miner *storage.Miner,
sealer sectorstorage.SectorManager,
full lapi.FullNode,
full v1api.FullNode,
ds dtypes.MetadataDS,
pieceStore dtypes.ProviderPieceStore,
mds dtypes.StagingMultiDstore,
@@ -678,7 +679,7 @@ func SectorStorage(mctx helpers.MetricsCtx, lc fx.Lifecycle, ls stores.LocalStor
return sst, nil
}
func StorageAuth(ctx helpers.MetricsCtx, ca lapi.Common) (sectorstorage.StorageAuth, error) {
func StorageAuth(ctx helpers.MetricsCtx, ca v0api.Common) (sectorstorage.StorageAuth, error) {
token, err := ca.AuthNew(ctx, []auth.Permission{"admin"})
if err != nil {
return nil, xerrors.Errorf("creating storage auth header: %w", err)
node/node_test.go
+30 -3
View File
@@ -58,9 +58,23 @@ 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 TestBatchDealInput(t *testing.T) {
logging.SetLogLevel("miner", "ERROR")
logging.SetLogLevel("chainstore", "ERROR")
logging.SetLogLevel("chain", "ERROR")
logging.SetLogLevel("sub", "ERROR")
logging.SetLogLevel("storageminer", "ERROR")
blockTime := 10 * time.Millisecond
// For these tests where the block time is artificially short, just use
// a deal start epoch that is guaranteed to be far enough in the future
// so that the deal starts sealing in time
dealStartEpoch := abi.ChainEpoch(2 << 12)
test.TestBatchDealInput(t, builder.MockSbBuilder, blockTime, dealStartEpoch)
}
func TestAPIDealFlowReal(t *testing.T) {
@@ -232,3 +246,16 @@ func TestWindowPostDisputeFails(t *testing.T) {
test.TestWindowPostDisputeFails(t, builder.MockSbBuilder, 2*time.Millisecond)
}
func TestDeadlineToggling(t *testing.T) {
if os.Getenv("LOTUS_TEST_DEADLINE_TOGGLING") != "1" {
t.Skip("this takes a few minutes, set LOTUS_TEST_DEADLINE_TOGGLING=1 to run")
}
logging.SetLogLevel("miner", "ERROR")
logging.SetLogLevel("chainstore", "ERROR")
logging.SetLogLevel("chain", "ERROR")
logging.SetLogLevel("sub", "ERROR")
logging.SetLogLevel("storageminer", "FATAL")
test.TestDeadlineToggling(t, builder.MockSbBuilder, 2*time.Millisecond)
}
node/test/builder.go
+77 -9
View File
@@ -12,20 +12,25 @@ import (
"testing"
"time"
"github.com/gorilla/mux"
"golang.org/x/xerrors"
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/go-jsonrpc"
"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/go-storedcounter"
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/api/client"
"github.com/filecoin-project/lotus/api/test"
"github.com/filecoin-project/lotus/api/v0api"
"github.com/filecoin-project/lotus/api/v1api"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain"
"github.com/filecoin-project/lotus/chain/actors"
"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
"github.com/filecoin-project/lotus/chain/actors/builtin/power"
"github.com/filecoin-project/lotus/chain/gen"
genesis2 "github.com/filecoin-project/lotus/chain/gen/genesis"
"github.com/filecoin-project/lotus/chain/messagepool"
@@ -44,6 +49,7 @@ import (
"github.com/filecoin-project/lotus/node/repo"
"github.com/filecoin-project/lotus/storage/mockstorage"
miner2 "github.com/filecoin-project/specs-actors/v2/actors/builtin/miner"
power2 "github.com/filecoin-project/specs-actors/v2/actors/builtin/power"
"github.com/ipfs/go-datastore"
"github.com/libp2p/go-libp2p-core/crypto"
"github.com/libp2p/go-libp2p-core/peer"
@@ -122,7 +128,7 @@ func CreateTestStorageNode(ctx context.Context, t *testing.T, waddr address.Addr
node.MockHost(mn),
node.Override(new(api.FullNode), tnd),
node.Override(new(v1api.FullNode), tnd),
node.Override(new(*lotusminer.Miner), lotusminer.NewTestMiner(mineBlock, act)),
opts,
@@ -151,6 +157,49 @@ func CreateTestStorageNode(ctx context.Context, t *testing.T, waddr address.Addr
return test.TestStorageNode{StorageMiner: minerapi, MineOne: mineOne, Stop: stop}
}
func storageBuilder(parentNode test.TestNode, mn mocknet.Mocknet, opts node.Option) test.StorageBuilder {
return func(ctx context.Context, t *testing.T, spt abi.RegisteredSealProof, owner address.Address) test.TestStorageNode {
pk, _, err := crypto.GenerateEd25519Key(rand.Reader)
require.NoError(t, err)
minerPid, err := peer.IDFromPrivateKey(pk)
require.NoError(t, err)
params, serr := actors.SerializeParams(&power2.CreateMinerParams{
Owner: owner,
Worker: owner,
SealProofType: spt,
Peer: abi.PeerID(minerPid),
})
require.NoError(t, serr)
createStorageMinerMsg := &types.Message{
To: power.Address,
From: owner,
Value: big.Zero(),
Method: power.Methods.CreateMiner,
Params: params,
GasLimit: 0,
GasPremium: big.NewInt(5252),
}
signed, err := parentNode.MpoolPushMessage(ctx, createStorageMinerMsg, nil)
require.NoError(t, err)
mw, err := parentNode.StateWaitMsg(ctx, signed.Cid(), build.MessageConfidence, api.LookbackNoLimit, true)
require.NoError(t, err)
require.Equal(t, exitcode.Ok, mw.Receipt.ExitCode)
var retval power2.CreateMinerReturn
err = retval.UnmarshalCBOR(bytes.NewReader(mw.Receipt.Return))
require.NoError(t, err)
return CreateTestStorageNode(ctx, t, owner, retval.IDAddress, pk, parentNode, mn, opts)
}
}
func Builder(t *testing.T, fullOpts []test.FullNodeOpts, storage []test.StorageMiner) ([]test.TestNode, []test.TestStorageNode) {
return mockBuilderOpts(t, fullOpts, storage, false)
}
@@ -266,6 +315,8 @@ func mockBuilderOpts(t *testing.T, fullOpts []test.FullNodeOpts, storage []test.
if rpc {
fulls[i] = fullRpc(t, fulls[i])
}
fulls[i].Stb = storageBuilder(fulls[i], mn, node.Options())
}
for i, def := range storage {
@@ -432,6 +483,14 @@ func mockSbBuilderOpts(t *testing.T, fullOpts []test.FullNodeOpts, storage []tes
if rpc {
fulls[i] = fullRpc(t, fulls[i])
}
fulls[i].Stb = storageBuilder(fulls[i], mn, node.Options(
node.Override(new(sectorstorage.SectorManager), func() (sectorstorage.SectorManager, error) {
return mock.NewMockSectorMgr(nil), nil
}),
node.Override(new(ffiwrapper.Verifier), mock.MockVerifier),
node.Unset(new(*sectorstorage.Manager)),
))
}
for i, def := range storage {
@@ -500,12 +559,15 @@ func mockSbBuilderOpts(t *testing.T, fullOpts []test.FullNodeOpts, storage []tes
}
func fullRpc(t *testing.T, nd test.TestNode) test.TestNode {
ma, listenAddr, err := CreateRPCServer(t, nd)
ma, listenAddr, err := CreateRPCServer(t, map[string]interface{}{
"/rpc/v1": nd,
"/rpc/v0": &v0api.WrapperV1Full{FullNode: nd},
})
require.NoError(t, err)
var stop func()
var full test.TestNode
full.FullNode, stop, err = client.NewFullNodeRPC(context.Background(), listenAddr, nil)
full.FullNode, stop, err = client.NewFullNodeRPCV1(context.Background(), listenAddr+"/rpc/v1", nil)
require.NoError(t, err)
t.Cleanup(stop)
@@ -514,12 +576,14 @@ func fullRpc(t *testing.T, nd test.TestNode) test.TestNode {
}
func storerRpc(t *testing.T, nd test.TestStorageNode) test.TestStorageNode {
ma, listenAddr, err := CreateRPCServer(t, nd)
ma, listenAddr, err := CreateRPCServer(t, map[string]interface{}{
"/rpc/v0": nd,
})
require.NoError(t, err)
var stop func()
var storer test.TestStorageNode
storer.StorageMiner, stop, err = client.NewStorageMinerRPC(context.Background(), listenAddr, nil)
storer.StorageMiner, stop, err = client.NewStorageMinerRPCV0(context.Background(), listenAddr+"/rpc/v0", nil)
require.NoError(t, err)
t.Cleanup(stop)
@@ -528,10 +592,14 @@ func storerRpc(t *testing.T, nd test.TestStorageNode) test.TestStorageNode {
return storer
}
func CreateRPCServer(t *testing.T, handler interface{}) (multiaddr.Multiaddr, string, error) {
rpcServer := jsonrpc.NewServer()
rpcServer.Register("Filecoin", handler)
testServ := httptest.NewServer(rpcServer) // todo: close
func CreateRPCServer(t *testing.T, handlers map[string]interface{}) (multiaddr.Multiaddr, string, error) {
m := mux.NewRouter()
for path, handler := range handlers {
rpcServer := jsonrpc.NewServer()
rpcServer.Register("Filecoin", handler)
m.Handle(path, rpcServer)
}
testServ := httptest.NewServer(m) // todo: close
t.Cleanup(testServ.Close)
t.Cleanup(testServ.CloseClientConnections)