Merge master into feat/nv13

node/builder.go

@@ -333,6 +333,7 @@ var ChainNode = Options(
 
 	// Lite node API
 	ApplyIf(isLiteNode,
+		Override(new(messagepool.Provider), messagepool.NewProviderLite),
 		Override(new(messagesigner.MpoolNonceAPI), From(new(modules.MpoolNonceAPI))),
 		Override(new(full.ChainModuleAPI), From(new(api.Gateway))),
 		Override(new(full.GasModuleAPI), From(new(api.Gateway))),
@@ -343,6 +344,7 @@ var ChainNode = Options(
 
 	// Full node API / service startup
 	ApplyIf(isFullNode,
+		Override(new(messagepool.Provider), messagepool.NewProvider),
 		Override(new(messagesigner.MpoolNonceAPI), From(new(*messagepool.MessagePool))),
 		Override(new(full.ChainModuleAPI), From(new(full.ChainModule))),
 		Override(new(full.GasModuleAPI), From(new(full.GasModule))),

node/config/def.go

@@ -231,7 +231,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

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"io"
 	"os"
+	"time"
 
 	"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
 
@@ -31,10 +32,12 @@ import (
 	"github.com/ipld/go-ipld-prime/traversal/selector/builder"
 	"github.com/libp2p/go-libp2p-core/host"
 	"github.com/libp2p/go-libp2p-core/peer"
+	"github.com/multiformats/go-multibase"
 	mh "github.com/multiformats/go-multihash"
 	"go.uber.org/fx"
 
 	"github.com/filecoin-project/go-address"
+	cborutil "github.com/filecoin-project/go-cbor-util"
 	"github.com/filecoin-project/go-commp-utils/ffiwrapper"
 	"github.com/filecoin-project/go-commp-utils/writer"
 	datatransfer "github.com/filecoin-project/go-data-transfer"
@@ -43,8 +46,10 @@ import (
 	rm "github.com/filecoin-project/go-fil-markets/retrievalmarket"
 	"github.com/filecoin-project/go-fil-markets/shared"
 	"github.com/filecoin-project/go-fil-markets/storagemarket"
+	"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/specs-actors/v3/actors/builtin/market"
 
 	marketevents "github.com/filecoin-project/lotus/markets/loggers"
 
@@ -91,7 +96,13 @@ func calcDealExpiration(minDuration uint64, md *dline.Info, startEpoch abi.Chain
 	minExp := startEpoch + abi.ChainEpoch(minDuration)
 
 	// Align on miners ProvingPeriodBoundary
-	return minExp + md.WPoStProvingPeriod - (minExp % md.WPoStProvingPeriod) + (md.PeriodStart % md.WPoStProvingPeriod) - 1
+	exp := minExp + md.WPoStProvingPeriod - (minExp % md.WPoStProvingPeriod) + (md.PeriodStart % md.WPoStProvingPeriod) - 1
+	// Should only be possible for miners created around genesis
+	for exp < minExp {
+		exp += md.WPoStProvingPeriod
+	}
+
+	return exp
 }
 
 func (a *API) imgr() *importmgr.Mgr {
@@ -99,8 +110,23 @@ func (a *API) imgr() *importmgr.Mgr {
 }
 
 func (a *API) ClientStartDeal(ctx context.Context, params *api.StartDealParams) (*cid.Cid, error) {
+	return a.dealStarter(ctx, params, false)
+}
+
+func (a *API) ClientStatelessDeal(ctx context.Context, params *api.StartDealParams) (*cid.Cid, error) {
+	return a.dealStarter(ctx, params, true)
+}
+
+func (a *API) dealStarter(ctx context.Context, params *api.StartDealParams, isStateless bool) (*cid.Cid, error) {
 	var storeID *multistore.StoreID
-	if params.Data.TransferType == storagemarket.TTGraphsync {
+	if isStateless {
+		if params.Data.TransferType != storagemarket.TTManual {
+			return nil, xerrors.Errorf("invalid transfer type %s for stateless storage deal", params.Data.TransferType)
+		}
+		if !params.EpochPrice.IsZero() {
+			return nil, xerrors.New("stateless storage deals can only be initiated with storage price of 0")
+		}
+	} else if params.Data.TransferType == storagemarket.TTGraphsync {
 		importIDs := a.imgr().List()
 		for _, importID := range importIDs {
 			info, err := a.imgr().Info(importID)
@@ -148,8 +174,6 @@ func (a *API) ClientStartDeal(ctx context.Context, params *api.StartDealParams)
 		return nil, xerrors.New("data doesn't fit in a sector")
 	}
 
-	providerInfo := utils.NewStorageProviderInfo(params.Miner, mi.Worker, mi.SectorSize, *mi.PeerId, mi.Multiaddrs)
-
 	dealStart := params.DealStartEpoch
 	if dealStart <= 0 { // unset, or explicitly 'epoch undefined'
 		ts, err := a.ChainHead(ctx)
@@ -171,25 +195,112 @@ func (a *API) ClientStartDeal(ctx context.Context, params *api.StartDealParams)
 		return nil, xerrors.Errorf("failed to get seal proof type: %w", err)
 	}
 
-	result, err := a.SMDealClient.ProposeStorageDeal(ctx, storagemarket.ProposeStorageDealParams{
-		Addr:          params.Wallet,
-		Info:          &providerInfo,
-		Data:          params.Data,
-		StartEpoch:    dealStart,
-		EndEpoch:      calcDealExpiration(params.MinBlocksDuration, md, dealStart),
-		Price:         params.EpochPrice,
-		Collateral:    params.ProviderCollateral,
-		Rt:            st,
-		FastRetrieval: params.FastRetrieval,
-		VerifiedDeal:  params.VerifiedDeal,
-		StoreID:       storeID,
-	})
+	// regular flow
+	if !isStateless {
+		providerInfo := utils.NewStorageProviderInfo(params.Miner, mi.Worker, mi.SectorSize, *mi.PeerId, mi.Multiaddrs)
 
-	if err != nil {
-		return nil, xerrors.Errorf("failed to start deal: %w", err)
+		result, err := a.SMDealClient.ProposeStorageDeal(ctx, storagemarket.ProposeStorageDealParams{
+			Addr:          params.Wallet,
+			Info:          &providerInfo,
+			Data:          params.Data,
+			StartEpoch:    dealStart,
+			EndEpoch:      calcDealExpiration(params.MinBlocksDuration, md, dealStart),
+			Price:         params.EpochPrice,
+			Collateral:    params.ProviderCollateral,
+			Rt:            st,
+			FastRetrieval: params.FastRetrieval,
+			VerifiedDeal:  params.VerifiedDeal,
+			StoreID:       storeID,
+		})
+
+		if err != nil {
+			return nil, xerrors.Errorf("failed to start deal: %w", err)
+		}
+
+		return &result.ProposalCid, nil
 	}
 
-	return &result.ProposalCid, nil
+	//
+	// stateless flow from here to the end
+	//
+
+	dealProposal := &market.DealProposal{
+		PieceCID:             *params.Data.PieceCid,
+		PieceSize:            params.Data.PieceSize.Padded(),
+		Client:               walletKey,
+		Provider:             params.Miner,
+		Label:                params.Data.Root.Encode(multibase.MustNewEncoder('u')),
+		StartEpoch:           dealStart,
+		EndEpoch:             calcDealExpiration(params.MinBlocksDuration, md, dealStart),
+		StoragePricePerEpoch: big.Zero(),
+		ProviderCollateral:   params.ProviderCollateral,
+		ClientCollateral:     big.Zero(),
+		VerifiedDeal:         params.VerifiedDeal,
+	}
+
+	if dealProposal.ProviderCollateral.IsZero() {
+		networkCollateral, err := a.StateDealProviderCollateralBounds(ctx, params.Data.PieceSize.Padded(), params.VerifiedDeal, types.EmptyTSK)
+		if err != nil {
+			return nil, xerrors.Errorf("failed to determine minimum provider collateral: %w", err)
+		}
+		dealProposal.ProviderCollateral = networkCollateral.Min
+	}
+
+	dealProposalSerialized, err := cborutil.Dump(dealProposal)
+	if err != nil {
+		return nil, xerrors.Errorf("failed to serialize deal proposal: %w", err)
+	}
+
+	dealProposalSig, err := a.WalletSign(ctx, walletKey, dealProposalSerialized)
+	if err != nil {
+		return nil, xerrors.Errorf("failed to sign proposal : %w", err)
+	}
+
+	dealProposalSigned := &market.ClientDealProposal{
+		Proposal:        *dealProposal,
+		ClientSignature: *dealProposalSig,
+	}
+	dStream, err := network.NewFromLibp2pHost(a.Host,
+		// params duplicated from .../node/modules/client.go
+		// https://github.com/filecoin-project/lotus/pull/5961#discussion_r629768011
+		network.RetryParameters(time.Second, 5*time.Minute, 15, 5),
+	).NewDealStream(ctx, *mi.PeerId)
+	if err != nil {
+		return nil, xerrors.Errorf("opening dealstream to %s/%s failed: %w", params.Miner, *mi.PeerId, err)
+	}
+
+	if err = dStream.WriteDealProposal(network.Proposal{
+		FastRetrieval: true,
+		DealProposal:  dealProposalSigned,
+		Piece: &storagemarket.DataRef{
+			TransferType: storagemarket.TTManual,
+			Root:         params.Data.Root,
+			PieceCid:     params.Data.PieceCid,
+			PieceSize:    params.Data.PieceSize,
+		},
+	}); err != nil {
+		return nil, xerrors.Errorf("sending deal proposal failed: %w", err)
+	}
+
+	resp, _, err := dStream.ReadDealResponse()
+	if err != nil {
+		return nil, xerrors.Errorf("reading proposal response failed: %w", err)
+	}
+
+	dealProposalIpld, err := cborutil.AsIpld(dealProposalSigned)
+	if err != nil {
+		return nil, xerrors.Errorf("serializing proposal node failed: %w", err)
+	}
+
+	if !dealProposalIpld.Cid().Equals(resp.Response.Proposal) {
+		return nil, xerrors.Errorf("provider returned proposal cid %s but we expected %s", resp.Response.Proposal, dealProposalIpld.Cid())
+	}
+
+	if resp.Response.State != storagemarket.StorageDealWaitingForData {
+		return nil, xerrors.Errorf("provider returned unexpected state %d for proposal %s, with message: %s", resp.Response.State, resp.Response.Proposal, resp.Response.Message)
+	}
+
+	return &resp.Response.Proposal, nil
 }
 
 func (a *API) ClientListDeals(ctx context.Context) ([]api.DealInfo, error) {
@@ -600,6 +711,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.go

@@ -2,16 +2,21 @@ package impl
 
 import (
 	"context"
+	"time"
+
+	"github.com/libp2p/go-libp2p-core/peer"
 
 	logging "github.com/ipfs/go-log/v2"
 
 	"github.com/filecoin-project/lotus/api"
+	"github.com/filecoin-project/lotus/build"
 	"github.com/filecoin-project/lotus/node/impl/client"
 	"github.com/filecoin-project/lotus/node/impl/common"
 	"github.com/filecoin-project/lotus/node/impl/full"
 	"github.com/filecoin-project/lotus/node/impl/market"
 	"github.com/filecoin-project/lotus/node/impl/paych"
 	"github.com/filecoin-project/lotus/node/modules/dtypes"
+	"github.com/filecoin-project/lotus/node/modules/lp2p"
 )
 
 var log = logging.Logger("node")
@@ -30,11 +35,86 @@ type FullNodeAPI struct {
 	full.SyncAPI
 	full.BeaconAPI
 
-	DS dtypes.MetadataDS
+	DS          dtypes.MetadataDS
+	NetworkName dtypes.NetworkName
 }
 
 func (n *FullNodeAPI) CreateBackup(ctx context.Context, fpath string) error {
 	return backup(n.DS, fpath)
 }
 
+func (n *FullNodeAPI) NodeStatus(ctx context.Context, inclChainStatus bool) (status api.NodeStatus, err error) {
+	curTs, err := n.ChainHead(ctx)
+	if err != nil {
+		return status, err
+	}
+
+	status.SyncStatus.Epoch = uint64(curTs.Height())
+	timestamp := time.Unix(int64(curTs.MinTimestamp()), 0)
+	delta := time.Since(timestamp).Seconds()
+	status.SyncStatus.Behind = uint64(delta / 30)
+
+	// get peers in the messages and blocks topics
+	peersMsgs := make(map[peer.ID]struct{})
+	peersBlocks := make(map[peer.ID]struct{})
+
+	for _, p := range n.PubSub.ListPeers(build.MessagesTopic(n.NetworkName)) {
+		peersMsgs[p] = struct{}{}
+	}
+
+	for _, p := range n.PubSub.ListPeers(build.BlocksTopic(n.NetworkName)) {
+		peersBlocks[p] = struct{}{}
+	}
+
+	// get scores for all connected and recent peers
+	scores, err := n.NetPubsubScores(ctx)
+	if err != nil {
+		return status, err
+	}
+
+	for _, score := range scores {
+		if score.Score.Score > lp2p.PublishScoreThreshold {
+			_, inMsgs := peersMsgs[score.ID]
+			if inMsgs {
+				status.PeerStatus.PeersToPublishMsgs++
+			}
+
+			_, inBlocks := peersBlocks[score.ID]
+			if inBlocks {
+				status.PeerStatus.PeersToPublishBlocks++
+			}
+		}
+	}
+
+	if inclChainStatus && status.SyncStatus.Epoch > uint64(build.Finality) {
+		blockCnt := 0
+		ts := curTs
+
+		for i := 0; i < 100; i++ {
+			blockCnt += len(ts.Blocks())
+			tsk := ts.Parents()
+			ts, err = n.ChainGetTipSet(ctx, tsk)
+			if err != nil {
+				return status, err
+			}
+		}
+
+		status.ChainStatus.BlocksPerTipsetLast100 = float64(blockCnt) / 100
+
+		for i := 100; i < int(build.Finality); i++ {
+			blockCnt += len(ts.Blocks())
+			tsk := ts.Parents()
+			ts, err = n.ChainGetTipSet(ctx, tsk)
+			if err != nil {
+				return status, err
+			}
+		}
+
+		status.ChainStatus.BlocksPerTipsetLastFinality = float64(blockCnt) / float64(build.Finality)
+
+	}
+
+	return status, nil
+}
+
 var _ api.FullNode = &FullNodeAPI{}

node/impl/full/chain.go

@@ -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

@@ -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

@@ -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).
@@ -223,8 +225,20 @@ func (a *MpoolAPI) MpoolBatchPushMessage(ctx context.Context, msgs []*types.Mess
 	return smsgs, nil
 }
 
+func (a *MpoolAPI) MpoolCheckMessages(ctx context.Context, protos []*api.MessagePrototype) ([][]api.MessageCheckStatus, error) {
+	return a.Mpool.CheckMessages(protos)
+}
+
+func (a *MpoolAPI) MpoolCheckPendingMessages(ctx context.Context, from address.Address) ([][]api.MessageCheckStatus, error) {
+	return a.Mpool.CheckPendingMessages(from)
+}
+
+func (a *MpoolAPI) MpoolCheckReplaceMessages(ctx context.Context, msgs []*types.Message) ([][]api.MessageCheckStatus, error) {
+	return a.Mpool.CheckReplaceMessages(msgs)
+}
+
 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/multisig.go

@@ -14,7 +14,6 @@ import (
 
 	multisig2 "github.com/filecoin-project/specs-actors/v2/actors/builtin/multisig"
 
-	"github.com/ipfs/go-cid"
 	"go.uber.org/fx"
 	"golang.org/x/xerrors"
 )
@@ -37,134 +36,129 @@ func (a *MsigAPI) messageBuilder(ctx context.Context, from address.Address) (mul
 
 // TODO: remove gp (gasPrice) from arguments
 // TODO: Add "vesting start" to arguments.
-func (a *MsigAPI) MsigCreate(ctx context.Context, req uint64, addrs []address.Address, duration abi.ChainEpoch, val types.BigInt, src address.Address, gp types.BigInt) (cid.Cid, error) {
+func (a *MsigAPI) MsigCreate(ctx context.Context, req uint64, addrs []address.Address, duration abi.ChainEpoch, val types.BigInt, src address.Address, gp types.BigInt) (*api.MessagePrototype, error) {
 
 	mb, err := a.messageBuilder(ctx, src)
 	if err != nil {
-		return cid.Undef, err
+		return nil, err
 	}
 
 	msg, err := mb.Create(addrs, req, 0, duration, val)
 	if err != nil {
-		return cid.Undef, err
+		return nil, err
 	}
 
-	// send the message out to the network
-	smsg, err := a.MpoolAPI.MpoolPushMessage(ctx, msg, nil)
-	if err != nil {
-		return cid.Undef, err
-	}
-
-	return smsg.Cid(), nil
+	return &api.MessagePrototype{
+		Message:    *msg,
+		ValidNonce: false,
+	}, nil
 }
 
-func (a *MsigAPI) MsigPropose(ctx context.Context, msig address.Address, to address.Address, amt types.BigInt, src address.Address, method uint64, params []byte) (cid.Cid, error) {
+func (a *MsigAPI) MsigPropose(ctx context.Context, msig address.Address, to address.Address, amt types.BigInt, src address.Address, method uint64, params []byte) (*api.MessagePrototype, error) {
 
 	mb, err := a.messageBuilder(ctx, src)
 	if err != nil {
-		return cid.Undef, err
+		return nil, err
 	}
 
 	msg, err := mb.Propose(msig, to, amt, abi.MethodNum(method), params)
 	if err != nil {
-		return cid.Undef, xerrors.Errorf("failed to create proposal: %w", err)
+		return nil, xerrors.Errorf("failed to create proposal: %w", err)
 	}
 
-	smsg, err := a.MpoolAPI.MpoolPushMessage(ctx, msg, nil)
-	if err != nil {
-		return cid.Undef, xerrors.Errorf("failed to push message: %w", err)
-	}
-
-	return smsg.Cid(), nil
+	return &api.MessagePrototype{
+		Message:    *msg,
+		ValidNonce: false,
+	}, nil
 }
 
-func (a *MsigAPI) MsigAddPropose(ctx context.Context, msig address.Address, src address.Address, newAdd address.Address, inc bool) (cid.Cid, error) {
+func (a *MsigAPI) MsigAddPropose(ctx context.Context, msig address.Address, src address.Address, newAdd address.Address, inc bool) (*api.MessagePrototype, error) {
 	enc, actErr := serializeAddParams(newAdd, inc)
 	if actErr != nil {
-		return cid.Undef, actErr
+		return nil, actErr
 	}
 
 	return a.MsigPropose(ctx, msig, msig, big.Zero(), src, uint64(multisig.Methods.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) {
+func (a *MsigAPI) MsigAddApprove(ctx context.Context, msig address.Address, src address.Address, txID uint64, proposer address.Address, newAdd address.Address, inc bool) (*api.MessagePrototype, error) {
 	enc, actErr := serializeAddParams(newAdd, inc)
 	if actErr != nil {
-		return cid.Undef, actErr
+		return nil, actErr
 	}
 
 	return a.MsigApproveTxnHash(ctx, msig, txID, proposer, msig, big.Zero(), src, uint64(multisig.Methods.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) {
+func (a *MsigAPI) MsigAddCancel(ctx context.Context, msig address.Address, src address.Address, txID uint64, newAdd address.Address, inc bool) (*api.MessagePrototype, error) {
 	enc, actErr := serializeAddParams(newAdd, inc)
 	if actErr != nil {
-		return cid.Undef, actErr
+		return nil, actErr
 	}
 
 	return a.MsigCancel(ctx, msig, txID, msig, big.Zero(), src, uint64(multisig.Methods.AddSigner), enc)
 }
 
-func (a *MsigAPI) MsigSwapPropose(ctx context.Context, msig address.Address, src address.Address, oldAdd address.Address, newAdd address.Address) (cid.Cid, error) {
+func (a *MsigAPI) MsigSwapPropose(ctx context.Context, msig address.Address, src address.Address, oldAdd address.Address, newAdd address.Address) (*api.MessagePrototype, error) {
 	enc, actErr := serializeSwapParams(oldAdd, newAdd)
 	if actErr != nil {
-		return cid.Undef, actErr
+		return nil, actErr
 	}
 
 	return a.MsigPropose(ctx, msig, msig, big.Zero(), src, uint64(multisig.Methods.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) {
+func (a *MsigAPI) MsigSwapApprove(ctx context.Context, msig address.Address, src address.Address, txID uint64, proposer address.Address, oldAdd address.Address, newAdd address.Address) (*api.MessagePrototype, error) {
 	enc, actErr := serializeSwapParams(oldAdd, newAdd)
 	if actErr != nil {
-		return cid.Undef, actErr
+		return nil, actErr
 	}
 
 	return a.MsigApproveTxnHash(ctx, msig, txID, proposer, msig, big.Zero(), src, uint64(multisig.Methods.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) {
+func (a *MsigAPI) MsigSwapCancel(ctx context.Context, msig address.Address, src address.Address, txID uint64, oldAdd address.Address, newAdd address.Address) (*api.MessagePrototype, error) {
 	enc, actErr := serializeSwapParams(oldAdd, newAdd)
 	if actErr != nil {
-		return cid.Undef, actErr
+		return nil, actErr
 	}
 
 	return a.MsigCancel(ctx, msig, txID, msig, big.Zero(), src, uint64(multisig.Methods.SwapSigner), enc)
 }
 
-func (a *MsigAPI) MsigApprove(ctx context.Context, msig address.Address, txID uint64, src address.Address) (cid.Cid, error) {
+func (a *MsigAPI) MsigApprove(ctx context.Context, msig address.Address, txID uint64, src address.Address) (*api.MessagePrototype, error) {
 	return a.msigApproveOrCancelSimple(ctx, api.MsigApprove, msig, txID, src)
 }
 
-func (a *MsigAPI) MsigApproveTxnHash(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) {
+func (a *MsigAPI) MsigApproveTxnHash(ctx context.Context, msig address.Address, txID uint64, proposer address.Address, to address.Address, amt types.BigInt, src address.Address, method uint64, params []byte) (*api.MessagePrototype, error) {
 	return a.msigApproveOrCancelTxnHash(ctx, api.MsigApprove, msig, txID, proposer, to, amt, src, method, params)
 }
 
-func (a *MsigAPI) MsigCancel(ctx context.Context, msig address.Address, txID uint64, to address.Address, amt types.BigInt, src address.Address, method uint64, params []byte) (cid.Cid, error) {
+func (a *MsigAPI) MsigCancel(ctx context.Context, msig address.Address, txID uint64, to address.Address, amt types.BigInt, src address.Address, method uint64, params []byte) (*api.MessagePrototype, error) {
 	return a.msigApproveOrCancelTxnHash(ctx, api.MsigCancel, msig, txID, src, to, amt, src, method, params)
 }
 
-func (a *MsigAPI) MsigRemoveSigner(ctx context.Context, msig address.Address, proposer address.Address, toRemove address.Address, decrease bool) (cid.Cid, error) {
+func (a *MsigAPI) MsigRemoveSigner(ctx context.Context, msig address.Address, proposer address.Address, toRemove address.Address, decrease bool) (*api.MessagePrototype, error) {
 	enc, actErr := serializeRemoveParams(toRemove, decrease)
 	if actErr != nil {
-		return cid.Undef, actErr
+		return nil, actErr
 	}
 
 	return a.MsigPropose(ctx, msig, msig, types.NewInt(0), proposer, uint64(multisig.Methods.RemoveSigner), enc)
 }
 
-func (a *MsigAPI) msigApproveOrCancelSimple(ctx context.Context, operation api.MsigProposeResponse, msig address.Address, txID uint64, src address.Address) (cid.Cid, error) {
+func (a *MsigAPI) msigApproveOrCancelSimple(ctx context.Context, operation api.MsigProposeResponse, msig address.Address, txID uint64, src address.Address) (*api.MessagePrototype, error) {
 	if msig == address.Undef {
-		return cid.Undef, xerrors.Errorf("must provide multisig address")
+		return nil, xerrors.Errorf("must provide multisig address")
 	}
 
 	if src == address.Undef {
-		return cid.Undef, xerrors.Errorf("must provide source address")
+		return nil, xerrors.Errorf("must provide source address")
 	}
 
 	mb, err := a.messageBuilder(ctx, src)
 	if err != nil {
-		return cid.Undef, err
+		return nil, err
 	}
 
 	var msg *types.Message
@@ -174,34 +168,31 @@ func (a *MsigAPI) msigApproveOrCancelSimple(ctx context.Context, operation api.M
 	case api.MsigCancel:
 		msg, err = mb.Cancel(msig, txID, nil)
 	default:
-		return cid.Undef, xerrors.Errorf("Invalid operation for msigApproveOrCancel")
+		return nil, xerrors.Errorf("Invalid operation for msigApproveOrCancel")
 	}
 	if err != nil {
-		return cid.Undef, err
+		return nil, err
 	}
 
-	smsg, err := a.MpoolAPI.MpoolPushMessage(ctx, msg, nil)
-	if err != nil {
-		return cid.Undef, err
-	}
-
-	return smsg.Cid(), nil
-
+	return &api.MessagePrototype{
+		Message:    *msg,
+		ValidNonce: false,
+	}, nil
 }
 
-func (a *MsigAPI) msigApproveOrCancelTxnHash(ctx context.Context, operation api.MsigProposeResponse, msig address.Address, txID uint64, proposer address.Address, to address.Address, amt types.BigInt, src address.Address, method uint64, params []byte) (cid.Cid, error) {
+func (a *MsigAPI) msigApproveOrCancelTxnHash(ctx context.Context, operation api.MsigProposeResponse, msig address.Address, txID uint64, proposer address.Address, to address.Address, amt types.BigInt, src address.Address, method uint64, params []byte) (*api.MessagePrototype, error) {
 	if msig == address.Undef {
-		return cid.Undef, xerrors.Errorf("must provide multisig address")
+		return nil, xerrors.Errorf("must provide multisig address")
 	}
 
 	if src == address.Undef {
-		return cid.Undef, xerrors.Errorf("must provide source address")
+		return nil, xerrors.Errorf("must provide source address")
 	}
 
 	if proposer.Protocol() != address.ID {
 		proposerID, err := a.StateAPI.StateLookupID(ctx, proposer, types.EmptyTSK)
 		if err != nil {
-			return cid.Undef, err
+			return nil, err
 		}
 		proposer = proposerID
 	}
@@ -216,7 +207,7 @@ func (a *MsigAPI) msigApproveOrCancelTxnHash(ctx context.Context, operation api.
 
 	mb, err := a.messageBuilder(ctx, src)
 	if err != nil {
-		return cid.Undef, err
+		return nil, err
 	}
 
 	var msg *types.Message
@@ -226,18 +217,16 @@ func (a *MsigAPI) msigApproveOrCancelTxnHash(ctx context.Context, operation api.
 	case api.MsigCancel:
 		msg, err = mb.Cancel(msig, txID, &p)
 	default:
-		return cid.Undef, xerrors.Errorf("Invalid operation for msigApproveOrCancel")
+		return nil, xerrors.Errorf("Invalid operation for msigApproveOrCancel")
 	}
 	if err != nil {
-		return cid.Undef, err
+		return nil, err
 	}
 
-	smsg, err := a.MpoolAPI.MpoolPushMessage(ctx, msg, nil)
-	if err != nil {
-		return cid.Undef, err
-	}
-
-	return smsg.Cid(), nil
+	return &api.MessagePrototype{
+		Message:    *msg,
+		ValidNonce: false,
+	}, nil
 }
 
 func serializeAddParams(new address.Address, inc bool) ([]byte, error) {

node/impl/full/state.go

@@ -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)
 		}
@@ -425,7 +426,7 @@ func (a *StateAPI) StateReplay(ctx context.Context, tsk types.TipSetKey, mc cid.
 	}, nil
 }
 
-func (m *StateModule) StateGetActor(ctx context.Context, actor address.Address, tsk types.TipSetKey) (*types.Actor, error) {
+func (m *StateModule) StateGetActor(ctx context.Context, actor address.Address, tsk types.TipSetKey) (a *types.Actor, err error) {
 	ts, err := m.Chain.GetTipSetFromKey(tsk)
 	if err != nil {
 		return nil, xerrors.Errorf("loading tipset %s: %w", tsk, 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

@@ -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

@@ -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

@@ -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/chain.go

@@ -9,7 +9,6 @@ import (
 	"github.com/ipfs/go-blockservice"
 	"github.com/libp2p/go-libp2p-core/host"
 	"github.com/libp2p/go-libp2p-core/routing"
-	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	"go.uber.org/fx"
 	"golang.org/x/xerrors"
 
@@ -59,8 +58,7 @@ func ChainBlockService(bs dtypes.ExposedBlockstore, rem dtypes.ChainBitswap) dty
 	return blockservice.New(bs, rem)
 }
 
-func MessagePool(lc fx.Lifecycle, sm *stmgr.StateManager, ps *pubsub.PubSub, ds dtypes.MetadataDS, nn dtypes.NetworkName, j journal.Journal) (*messagepool.MessagePool, error) {
-	mpp := messagepool.NewProvider(sm, ps)
+func MessagePool(lc fx.Lifecycle, mpp messagepool.Provider, ds dtypes.MetadataDS, nn dtypes.NetworkName, j journal.Journal) (*messagepool.MessagePool, error) {
 	mp, err := messagepool.New(mpp, ds, nn, j)
 	if err != nil {
 		return nil, xerrors.Errorf("constructing mpool: %w", err)

node/modules/client.go

@@ -134,26 +134,21 @@ func NewClientGraphsyncDataTransfer(lc fx.Lifecycle, h host.Host, gs dtypes.Grap
 
 	// data-transfer push / pull channel restart configuration:
 	dtRestartConfig := dtimpl.ChannelRestartConfig(channelmonitor.Config{
-		// For now only monitor push channels (for storage deals)
-		MonitorPushChannels: true,
-		// TODO: Enable pull channel monitoring (for retrievals) when the
-		//  following issue has been fixed:
-		// https://github.com/filecoin-project/go-data-transfer/issues/172
-		MonitorPullChannels: false,
-		// Wait up to 30s for the other side to respond to an Open channel message
-		AcceptTimeout: 30 * time.Second,
-		// Send a restart message if the data rate falls below 1024 bytes / minute
-		Interval:            time.Minute,
-		MinBytesTransferred: 1024,
-		// Perform check 10 times / minute
-		ChecksPerInterval: 10,
+		// Wait up to 2m for the other side to respond to an Open channel message
+		AcceptTimeout: 2 * time.Minute,
+		// When an error occurs, wait a little while until all related errors
+		// have fired before sending a restart message
+		RestartDebounce: 10 * time.Second,
 		// After sending a restart, wait for at least 1 minute before sending another
 		RestartBackoff: time.Minute,
 		// After trying to restart 3 times, give up and fail the transfer
 		MaxConsecutiveRestarts: 3,
-		// Wait up to 30s for the other side to send a Complete message once all
+		// After sending a restart message, the time to wait for the peer to
+		// respond with an ack of the restart
+		RestartAckTimeout: 30 * time.Second,
+		// Wait up to 10m for the other side to send a Complete message once all
 		// data has been sent / received
-		CompleteTimeout: 30 * time.Second,
+		CompleteTimeout: 10 * time.Minute,
 	})
 	dt, err := dtimpl.NewDataTransfer(dtDs, filepath.Join(r.Path(), "data-transfer"), net, transport, dtRestartConfig)
 	if err != nil {

node/modules/core.go

@@ -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/lp2p/pubsub.go

@@ -36,6 +36,15 @@ func init() {
 	pubsub.GossipSubHistoryLength = 10
 	pubsub.GossipSubGossipFactor = 0.1
 }
+
+const (
+	GossipScoreThreshold             = -500
+	PublishScoreThreshold            = -1000
+	GraylistScoreThreshold           = -2500
+	AcceptPXScoreThreshold           = 1000
+	OpportunisticGraftScoreThreshold = 3.5
+)
+
 func ScoreKeeper() *dtypes.ScoreKeeper {
 	return new(dtypes.ScoreKeeper)
 }
@@ -256,11 +265,11 @@ func GossipSub(in GossipIn) (service *pubsub.PubSub, err error) {
 				Topics: topicParams,
 			},
 			&pubsub.PeerScoreThresholds{
-				GossipThreshold:             -500,
-				PublishThreshold:            -1000,
-				GraylistThreshold:           -2500,
-				AcceptPXThreshold:           1000,
-				OpportunisticGraftThreshold: 3.5,
+				GossipThreshold:             GossipScoreThreshold,
+				PublishThreshold:            PublishScoreThreshold,
+				GraylistThreshold:           GraylistScoreThreshold,
+				AcceptPXThreshold:           AcceptPXScoreThreshold,
+				OpportunisticGraftThreshold: OpportunisticGraftScoreThreshold,
 			},
 		),
 		pubsub.WithPeerScoreInspect(in.Sk.Update, 10*time.Second),

node/modules/mpoolnonceapi.go

@@ -2,6 +2,7 @@ package modules
 
 import (
 	"context"
+	"strings"
 
 	"go.uber.org/fx"
 	"golang.org/x/xerrors"
@@ -19,44 +20,89 @@ 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, xerrors.Errorf("getting actor converted: %w", 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
 }
 
+func (a *MpoolNonceAPI) GetActor(ctx context.Context, addr address.Address, tsk types.TipSetKey) (*types.Actor, error) {
+	act, err := a.StateModule.StateGetActor(ctx, addr, tsk)
+	if err != nil {
+		return nil, xerrors.Errorf("calling StateGetActor: %w", err)
+	}
+
+	return act, nil
+}
+
 var _ messagesigner.MpoolNonceAPI = (*MpoolNonceAPI)(nil)

node/modules/paych.go

@@ -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

@@ -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
@@ -439,7 +440,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
@@ -462,7 +463,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 {
@@ -637,7 +638,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,
@@ -680,7 +681,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

@@ -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) {

node/test/builder.go

@@ -12,6 +12,9 @@ import (
 	"testing"
 	"time"
 
+	"github.com/filecoin-project/go-state-types/network"
+
+	"github.com/gorilla/mux"
 	"golang.org/x/xerrors"
 
 	"github.com/filecoin-project/go-address"
@@ -23,6 +26,8 @@ import (
 	"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"
@@ -125,7 +130,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,
@@ -185,7 +190,7 @@ func storageBuilder(parentNode test.TestNode, mn mocknet.Mocknet, opts node.Opti
 		signed, err := parentNode.MpoolPushMessage(ctx, createStorageMinerMsg, nil)
 		require.NoError(t, err)
 
-		mw, err := parentNode.StateWaitMsg(ctx, signed.Cid(), build.MessageConfidence)
+		mw, err := parentNode.StateWaitMsg(ctx, signed.Cid(), build.MessageConfidence, api.LookbackNoLimit, true)
 		require.NoError(t, err)
 		require.Equal(t, exitcode.Ok, mw.Receipt.ExitCode)
 
@@ -274,6 +279,7 @@ func mockBuilderOpts(t *testing.T, fullOpts []test.FullNodeOpts, storage []test.
 		genms = append(genms, *genm)
 	}
 	templ := &genesis.Template{
+		NetworkVersion:   network.Version0,
 		Accounts:         genaccs,
 		Miners:           genms,
 		NetworkName:      "test",
@@ -437,6 +443,7 @@ func mockSbBuilderOpts(t *testing.T, fullOpts []test.FullNodeOpts, storage []tes
 		genms = append(genms, *genm)
 	}
 	templ := &genesis.Template{
+		NetworkVersion:   network.Version0,
 		Accounts:         genaccs,
 		Miners:           genms,
 		NetworkName:      "test",
@@ -559,12 +566,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)
 
@@ -573,12 +583,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)
 
@@ -587,10 +599,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)