Merge remote-tracking branch 'origin/master' into feat/async-restartable-workers

node/builder.go

@@ -3,9 +3,15 @@ package node
 import (
 	"context"
 	"errors"
-	"os"
 	"time"
 
+	"github.com/filecoin-project/lotus/chain"
+	"github.com/filecoin-project/lotus/chain/exchange"
+	"github.com/filecoin-project/lotus/chain/store"
+	"github.com/filecoin-project/lotus/chain/vm"
+	"github.com/filecoin-project/lotus/chain/wallet"
+	"github.com/filecoin-project/lotus/node/hello"
+
 	logging "github.com/ipfs/go-log"
 	ci "github.com/libp2p/go-libp2p-core/crypto"
 	"github.com/libp2p/go-libp2p-core/host"
@@ -29,9 +35,7 @@ import (
 	storage2 "github.com/filecoin-project/specs-storage/storage"
 
 	"github.com/filecoin-project/lotus/api"
-	"github.com/filecoin-project/lotus/chain"
 	"github.com/filecoin-project/lotus/chain/beacon"
-	"github.com/filecoin-project/lotus/chain/exchange"
 	"github.com/filecoin-project/lotus/chain/gen"
 	"github.com/filecoin-project/lotus/chain/gen/slashfilter"
 	"github.com/filecoin-project/lotus/chain/market"
@@ -39,10 +43,9 @@ import (
 	"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"
 	"github.com/filecoin-project/lotus/chain/types"
-	"github.com/filecoin-project/lotus/chain/vm"
-	"github.com/filecoin-project/lotus/chain/wallet"
+	ledgerwallet "github.com/filecoin-project/lotus/chain/wallet/ledger"
+	"github.com/filecoin-project/lotus/chain/wallet/remotewallet"
 	sectorstorage "github.com/filecoin-project/lotus/extern/sector-storage"
 	"github.com/filecoin-project/lotus/extern/sector-storage/ffiwrapper"
 	"github.com/filecoin-project/lotus/extern/sector-storage/stores"
@@ -57,9 +60,9 @@ import (
 	"github.com/filecoin-project/lotus/markets/storageadapter"
 	"github.com/filecoin-project/lotus/miner"
 	"github.com/filecoin-project/lotus/node/config"
-	"github.com/filecoin-project/lotus/node/hello"
 	"github.com/filecoin-project/lotus/node/impl"
 	"github.com/filecoin-project/lotus/node/impl/common"
+	"github.com/filecoin-project/lotus/node/impl/full"
 	"github.com/filecoin-project/lotus/node/modules"
 	"github.com/filecoin-project/lotus/node/modules/dtypes"
 	"github.com/filecoin-project/lotus/node/modules/helpers"
@@ -72,10 +75,6 @@ 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")
 
@@ -159,25 +158,14 @@ type Settings struct {
 
 	Online bool // Online option applied
 	Config bool // Config option applied
-
+	Lite   bool // Start node in "lite" mode
 }
 
 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.DisabledEvents), journal.EnvDisabledEvents),
 		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),
@@ -233,6 +221,10 @@ func isType(t repo.RepoType) func(s *Settings) bool {
 
 // Online sets up basic libp2p node
 func Online() Option {
+	isFullOrLiteNode := func(s *Settings) bool { return s.nodeType == repo.FullNode }
+	isFullNode := func(s *Settings) bool { return s.nodeType == repo.FullNode && !s.Lite }
+	isLiteNode := func(s *Settings) bool { return s.nodeType == repo.FullNode && s.Lite }
+
 	return Options(
 		// make sure that online is applied before Config.
 		// This is important because Config overrides some of Online units
@@ -246,23 +238,23 @@ func Online() Option {
 		// common
 		Override(new(*slashfilter.SlashFilter), modules.NewSlashFilter),
 
-		// Full node
-
-		ApplyIf(isType(repo.FullNode),
+		// Full node or lite node
+		ApplyIf(isFullOrLiteNode,
 			// TODO: Fix offline mode
 
 			Override(new(dtypes.BootstrapPeers), modules.BuiltinBootstrap),
 			Override(new(dtypes.DrandBootstrap), modules.DrandBootstrap),
 			Override(new(dtypes.DrandSchedule), modules.BuiltinDrandConfig),
 
-			Override(HandleIncomingMessagesKey, modules.HandleIncomingMessages),
-
 			Override(new(ffiwrapper.Verifier), ffiwrapper.ProofVerifier),
 			Override(new(vm.SyscallBuilder), vm.Syscalls),
 			Override(new(*store.ChainStore), modules.ChainStore),
 			Override(new(stmgr.UpgradeSchedule), stmgr.DefaultUpgradeSchedule()),
 			Override(new(*stmgr.StateManager), stmgr.NewStateManagerWithUpgradeSchedule),
-			Override(new(*wallet.Wallet), wallet.NewWallet),
+			Override(new(stmgr.StateManagerAPI), From(new(*stmgr.StateManager))),
+			Override(new(*wallet.LocalWallet), wallet.NewWallet),
+			Override(new(wallet.Default), From(new(*wallet.LocalWallet))),
+			Override(new(api.WalletAPI), From(new(wallet.MultiWallet))),
 			Override(new(*messagesigner.MessageSigner), messagesigner.NewMessageSigner),
 
 			Override(new(dtypes.ChainGCLocker), blockstore.NewGCLocker),
@@ -289,12 +281,6 @@ func Online() Option {
 
 			Override(new(dtypes.Graphsync), modules.Graphsync),
 			Override(new(*dtypes.MpoolLocker), new(dtypes.MpoolLocker)),
-
-			Override(RunHelloKey, modules.RunHello),
-			Override(RunChainExchangeKey, modules.RunChainExchange),
-			Override(RunPeerMgrKey, modules.RunPeerMgr),
-			Override(HandleIncomingBlocksKey, modules.HandleIncomingBlocks),
-
 			Override(new(*discoveryimpl.Local), modules.NewLocalDiscovery),
 			Override(new(discovery.PeerResolver), modules.RetrievalResolver),
 
@@ -313,8 +299,34 @@ func Online() Option {
 			Override(SettlePaymentChannelsKey, settler.SettlePaymentChannels),
 		),
 
+		// Lite node
+		ApplyIf(isLiteNode,
+			Override(new(messagesigner.MpoolNonceAPI), From(new(modules.MpoolNonceAPI))),
+			Override(new(full.ChainModuleAPI), From(new(api.GatewayAPI))),
+			Override(new(full.GasModuleAPI), From(new(api.GatewayAPI))),
+			Override(new(full.MpoolModuleAPI), From(new(api.GatewayAPI))),
+			Override(new(full.StateModuleAPI), From(new(api.GatewayAPI))),
+			Override(new(stmgr.StateManagerAPI), modules.NewRPCStateManager),
+		),
+
+		// Full node
+		ApplyIf(isFullNode,
+			Override(new(messagesigner.MpoolNonceAPI), From(new(*messagepool.MessagePool))),
+			Override(new(full.ChainModuleAPI), From(new(full.ChainModule))),
+			Override(new(full.GasModuleAPI), From(new(full.GasModule))),
+			Override(new(full.MpoolModuleAPI), From(new(full.MpoolModule))),
+			Override(new(full.StateModuleAPI), From(new(full.StateModule))),
+			Override(new(stmgr.StateManagerAPI), From(new(*stmgr.StateManager))),
+
+			Override(RunHelloKey, modules.RunHello),
+			Override(RunChainExchangeKey, modules.RunChainExchange),
+			Override(RunPeerMgrKey, modules.RunPeerMgr),
+			Override(HandleIncomingMessagesKey, modules.HandleIncomingMessages),
+			Override(HandleIncomingBlocksKey, modules.HandleIncomingBlocks),
+		),
+
 		// miner
-		ApplyIf(func(s *Settings) bool { return s.nodeType == repo.StorageMiner },
+		ApplyIf(isType(repo.StorageMiner),
 			Override(new(api.Common), From(new(common.CommonAPI))),
 			Override(new(sectorstorage.StorageAuth), modules.StorageAuth),
 
@@ -450,6 +462,17 @@ func ConfigFullNode(c interface{}) Option {
 		If(cfg.Metrics.HeadNotifs,
 			Override(HeadMetricsKey, metrics.SendHeadNotifs(cfg.Metrics.Nickname)),
 		),
+
+		If(cfg.Wallet.RemoteBackend != "",
+			Override(new(*remotewallet.RemoteWallet), remotewallet.SetupRemoteWallet(cfg.Wallet.RemoteBackend)),
+		),
+		If(cfg.Wallet.EnableLedger,
+			Override(new(*ledgerwallet.LedgerWallet), ledgerwallet.NewWallet),
+		),
+		If(cfg.Wallet.DisableLocal,
+			Unset(new(*wallet.LocalWallet)),
+			Override(new(wallet.Default), wallet.NilDefault),
+		),
 	)
 }
 
@@ -509,12 +532,22 @@ func Repo(r repo.Repo) Option {
 	}
 }
 
-func FullAPI(out *api.FullNode) Option {
+type FullOption = Option
+
+func Lite(enable bool) FullOption {
+	return func(s *Settings) error {
+		s.Lite = enable
+		return nil
+	}
+}
+
+func FullAPI(out *api.FullNode, fopts ...FullOption) Option {
 	return Options(
 		func(s *Settings) error {
 			s.nodeType = repo.FullNode
 			return nil
 		},
+		Options(fopts...),
 		func(s *Settings) error {
 			resAPI := &impl.FullNodeAPI{}
 			s.invokes[ExtractApiKey] = fx.Populate(resAPI)

node/config/def.go

@@ -22,6 +22,7 @@ type FullNode struct {
 	Common
 	Client  Client
 	Metrics Metrics
+	Wallet  Wallet
 }
 
 // // Common
@@ -107,6 +108,12 @@ type Client struct {
 	IpfsUseForRetrieval bool
 }
 
+type Wallet struct {
+	RemoteBackend string
+	EnableLedger  bool
+	DisableLocal  bool
+}
+
 func defCommon() Common {
 	return Common{
 		API: API{

node/impl/client/client.go

@@ -462,13 +462,19 @@ type retrievalSubscribeEvent struct {
 	state rm.ClientDealState
 }
 
-func readSubscribeEvents(ctx context.Context, subscribeEvents chan retrievalSubscribeEvent, events chan marketevents.RetrievalEvent) error {
+func readSubscribeEvents(ctx context.Context, dealID retrievalmarket.DealID, subscribeEvents chan retrievalSubscribeEvent, events chan marketevents.RetrievalEvent) error {
 	for {
 		var subscribeEvent retrievalSubscribeEvent
 		select {
 		case <-ctx.Done():
 			return xerrors.New("Retrieval Timed Out")
 		case subscribeEvent = <-subscribeEvents:
+			if subscribeEvent.state.ID != dealID {
+				// we can't check the deal ID ahead of time because:
+				// 1. We need to subscribe before retrieving.
+				// 2. We won't know the deal ID until after retrieving.
+				continue
+			}
 		}
 
 		select {
@@ -531,19 +537,6 @@ func (a *API) clientRetrieve(ctx context.Context, order api.RetrievalOrder, ref
 		return err
 	}*/
 
-	var dealID retrievalmarket.DealID
-	subscribeEvents := make(chan retrievalSubscribeEvent, 1)
-	subscribeCtx, cancel := context.WithCancel(ctx)
-	defer cancel()
-	unsubscribe := a.Retrieval.SubscribeToEvents(func(event rm.ClientEvent, state rm.ClientDealState) {
-		if state.PayloadCID.Equals(order.Root) && state.ID == dealID {
-			select {
-			case <-subscribeCtx.Done():
-			case subscribeEvents <- retrievalSubscribeEvent{event, state}:
-			}
-		}
-	})
-
 	ppb := types.BigDiv(order.Total, types.NewInt(order.Size))
 
 	params, err := rm.NewParamsV1(ppb, order.PaymentInterval, order.PaymentIntervalIncrease, shared.AllSelector(), order.Piece, order.UnsealPrice)
@@ -562,7 +555,21 @@ func (a *API) clientRetrieve(ctx context.Context, order api.RetrievalOrder, ref
 		_ = a.RetrievalStoreMgr.ReleaseStore(store)
 	}()
 
-	dealID, err = a.Retrieval.Retrieve(
+	// Subscribe to events before retrieving to avoid losing events.
+	subscribeEvents := make(chan retrievalSubscribeEvent, 1)
+	subscribeCtx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	unsubscribe := a.Retrieval.SubscribeToEvents(func(event rm.ClientEvent, state rm.ClientDealState) {
+		// We'll check the deal IDs inside readSubscribeEvents.
+		if state.PayloadCID.Equals(order.Root) {
+			select {
+			case <-subscribeCtx.Done():
+			case subscribeEvents <- retrievalSubscribeEvent{event, state}:
+			}
+		}
+	})
+
+	dealID, err := a.Retrieval.Retrieve(
 		ctx,
 		order.Root,
 		params,
@@ -573,11 +580,12 @@ func (a *API) clientRetrieve(ctx context.Context, order api.RetrievalOrder, ref
 		store.StoreID())
 
 	if err != nil {
+		unsubscribe()
 		finish(xerrors.Errorf("Retrieve failed: %w", err))
 		return
 	}
 
-	err = readSubscribeEvents(ctx, subscribeEvents, events)
+	err = readSubscribeEvents(ctx, dealID, subscribeEvents, events)
 
 	unsubscribe()
 	if err != nil {

node/impl/full/chain.go

@@ -39,10 +39,28 @@ import (
 
 var log = logging.Logger("fullnode")
 
+type ChainModuleAPI interface {
+	ChainHead(context.Context) (*types.TipSet, error)
+	ChainGetTipSet(ctx context.Context, tsk types.TipSetKey) (*types.TipSet, error)
+	ChainGetTipSetByHeight(ctx context.Context, h abi.ChainEpoch, tsk types.TipSetKey) (*types.TipSet, error)
+}
+
+// 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).
+type ChainModule struct {
+	fx.In
+
+	Chain *store.ChainStore
+}
+
+var _ ChainModuleAPI = (*ChainModule)(nil)
+
 type ChainAPI struct {
 	fx.In
 
 	WalletAPI
+	ChainModuleAPI
 
 	Chain *store.ChainStore
 }
@@ -51,8 +69,8 @@ func (a *ChainAPI) ChainNotify(ctx context.Context) (<-chan []*api.HeadChange, e
 	return a.Chain.SubHeadChanges(ctx), nil
 }
 
-func (a *ChainAPI) ChainHead(context.Context) (*types.TipSet, error) {
-	return a.Chain.GetHeaviestTipSet(), nil
+func (m *ChainModule) ChainHead(context.Context) (*types.TipSet, error) {
+	return m.Chain.GetHeaviestTipSet(), nil
 }
 
 func (a *ChainAPI) ChainGetRandomnessFromTickets(ctx context.Context, tsk types.TipSetKey, personalization crypto.DomainSeparationTag, randEpoch abi.ChainEpoch, entropy []byte) (abi.Randomness, error) {
@@ -77,8 +95,8 @@ func (a *ChainAPI) ChainGetBlock(ctx context.Context, msg cid.Cid) (*types.Block
 	return a.Chain.GetBlock(msg)
 }
 
-func (a *ChainAPI) ChainGetTipSet(ctx context.Context, key types.TipSetKey) (*types.TipSet, error) {
-	return a.Chain.LoadTipSet(key)
+func (m *ChainModule) ChainGetTipSet(ctx context.Context, key types.TipSetKey) (*types.TipSet, error) {
+	return m.Chain.LoadTipSet(key)
 }
 
 func (a *ChainAPI) ChainGetBlockMessages(ctx context.Context, msg cid.Cid) (*api.BlockMessages, error) {
@@ -180,12 +198,12 @@ func (a *ChainAPI) ChainGetParentReceipts(ctx context.Context, bcid cid.Cid) ([]
 	return out, nil
 }
 
-func (a *ChainAPI) ChainGetTipSetByHeight(ctx context.Context, h abi.ChainEpoch, tsk types.TipSetKey) (*types.TipSet, error) {
-	ts, err := a.Chain.GetTipSetFromKey(tsk)
+func (m *ChainModule) ChainGetTipSetByHeight(ctx context.Context, h abi.ChainEpoch, tsk types.TipSetKey) (*types.TipSet, error) {
+	ts, err := m.Chain.GetTipSetFromKey(tsk)
 	if err != nil {
 		return nil, xerrors.Errorf("loading tipset %s: %w", tsk, err)
 	}
-	return a.Chain.GetTipsetByHeight(ctx, h, ts, true)
+	return m.Chain.GetTipsetByHeight(ctx, h, ts, true)
 }
 
 func (a *ChainAPI) ChainReadObj(ctx context.Context, obj cid.Cid) ([]byte, error) {

node/impl/full/gas.go

@@ -26,8 +26,27 @@ import (
 	"github.com/filecoin-project/lotus/chain/types"
 )
 
+type GasModuleAPI interface {
+	GasEstimateMessageGas(ctx context.Context, msg *types.Message, spec *api.MessageSendSpec, tsk types.TipSetKey) (*types.Message, error)
+}
+
+// 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).
+type GasModule struct {
+	fx.In
+	Stmgr *stmgr.StateManager
+	Chain *store.ChainStore
+	Mpool *messagepool.MessagePool
+}
+
+var _ GasModuleAPI = (*GasModule)(nil)
+
 type GasAPI struct {
 	fx.In
+
+	GasModuleAPI
+
 	Stmgr *stmgr.StateManager
 	Chain *store.ChainStore
 	Mpool *messagepool.MessagePool
@@ -36,9 +55,24 @@ type GasAPI struct {
 const MinGasPremium = 100e3
 const MaxSpendOnFeeDenom = 100
 
-func (a *GasAPI) GasEstimateFeeCap(ctx context.Context, msg *types.Message, maxqueueblks int64,
-	tsk types.TipSetKey) (types.BigInt, error) {
-	ts := a.Chain.GetHeaviestTipSet()
+func (a *GasAPI) GasEstimateFeeCap(
+	ctx context.Context,
+	msg *types.Message,
+	maxqueueblks int64,
+	tsk types.TipSetKey,
+) (types.BigInt, error) {
+	return gasEstimateFeeCap(a.Chain, msg, maxqueueblks)
+}
+func (m *GasModule) GasEstimateFeeCap(
+	ctx context.Context,
+	msg *types.Message,
+	maxqueueblks int64,
+	tsk types.TipSetKey,
+) (types.BigInt, error) {
+	return gasEstimateFeeCap(m.Chain, msg, maxqueueblks)
+}
+func gasEstimateFeeCap(cstore *store.ChainStore, msg *types.Message, maxqueueblks int64) (types.BigInt, error) {
+	ts := cstore.GetHeaviestTipSet()
 
 	parentBaseFee := ts.Blocks()[0].ParentBaseFee
 	increaseFactor := math.Pow(1.+1./float64(build.BaseFeeMaxChangeDenom), float64(maxqueueblks))
@@ -82,9 +116,25 @@ func medianGasPremium(prices []gasMeta, blocks int) abi.TokenAmount {
 	return premium
 }
 
-func (a *GasAPI) GasEstimateGasPremium(ctx context.Context, nblocksincl uint64,
-	sender address.Address, gaslimit int64, _ types.TipSetKey) (types.BigInt, error) {
-
+func (a *GasAPI) GasEstimateGasPremium(
+	ctx context.Context,
+	nblocksincl uint64,
+	sender address.Address,
+	gaslimit int64,
+	_ types.TipSetKey,
+) (types.BigInt, error) {
+	return gasEstimateGasPremium(a.Chain, nblocksincl)
+}
+func (m *GasModule) GasEstimateGasPremium(
+	ctx context.Context,
+	nblocksincl uint64,
+	sender address.Address,
+	gaslimit int64,
+	_ types.TipSetKey,
+) (types.BigInt, error) {
+	return gasEstimateGasPremium(m.Chain, nblocksincl)
+}
+func gasEstimateGasPremium(cstore *store.ChainStore, nblocksincl uint64) (types.BigInt, error) {
 	if nblocksincl == 0 {
 		nblocksincl = 1
 	}
@@ -92,20 +142,20 @@ func (a *GasAPI) GasEstimateGasPremium(ctx context.Context, nblocksincl uint64,
 	var prices []gasMeta
 	var blocks int
 
-	ts := a.Chain.GetHeaviestTipSet()
+	ts := cstore.GetHeaviestTipSet()
 	for i := uint64(0); i < nblocksincl*2; i++ {
 		if ts.Height() == 0 {
 			break // genesis
 		}
 
-		pts, err := a.Chain.LoadTipSet(ts.Parents())
+		pts, err := cstore.LoadTipSet(ts.Parents())
 		if err != nil {
 			return types.BigInt{}, err
 		}
 
 		blocks += len(pts.Blocks())
 
-		msgs, err := a.Chain.MessagesForTipset(pts)
+		msgs, err := cstore.MessagesForTipset(pts)
 		if err != nil {
 			return types.BigInt{}, xerrors.Errorf("loading messages: %w", err)
 		}
@@ -142,19 +192,30 @@ func (a *GasAPI) GasEstimateGasPremium(ctx context.Context, nblocksincl uint64,
 }
 
 func (a *GasAPI) GasEstimateGasLimit(ctx context.Context, msgIn *types.Message, _ types.TipSetKey) (int64, error) {
-
+	return gasEstimateGasLimit(ctx, a.Chain, a.Stmgr, a.Mpool, msgIn)
+}
+func (m *GasModule) GasEstimateGasLimit(ctx context.Context, msgIn *types.Message, _ types.TipSetKey) (int64, error) {
+	return gasEstimateGasLimit(ctx, m.Chain, m.Stmgr, m.Mpool, msgIn)
+}
+func gasEstimateGasLimit(
+	ctx context.Context,
+	cstore *store.ChainStore,
+	smgr *stmgr.StateManager,
+	mpool *messagepool.MessagePool,
+	msgIn *types.Message,
+) (int64, error) {
 	msg := *msgIn
 	msg.GasLimit = build.BlockGasLimit
 	msg.GasFeeCap = types.NewInt(uint64(build.MinimumBaseFee) + 1)
 	msg.GasPremium = types.NewInt(1)
 
-	currTs := a.Chain.GetHeaviestTipSet()
-	fromA, err := a.Stmgr.ResolveToKeyAddress(ctx, msgIn.From, currTs)
+	currTs := cstore.GetHeaviestTipSet()
+	fromA, err := smgr.ResolveToKeyAddress(ctx, msgIn.From, currTs)
 	if err != nil {
 		return -1, xerrors.Errorf("getting key address: %w", err)
 	}
 
-	pending, ts := a.Mpool.PendingFor(fromA)
+	pending, ts := mpool.PendingFor(fromA)
 	priorMsgs := make([]types.ChainMsg, 0, len(pending))
 	for _, m := range pending {
 		priorMsgs = append(priorMsgs, m)
@@ -163,11 +224,11 @@ func (a *GasAPI) GasEstimateGasLimit(ctx context.Context, msgIn *types.Message,
 	// Try calling until we find a height with no migration.
 	var res *api.InvocResult
 	for {
-		res, err = a.Stmgr.CallWithGas(ctx, &msg, priorMsgs, ts)
+		res, err = smgr.CallWithGas(ctx, &msg, priorMsgs, ts)
 		if err != stmgr.ErrExpensiveFork {
 			break
 		}
-		ts, err = a.Chain.GetTipSetFromKey(ts.Parents())
+		ts, err = cstore.GetTipSetFromKey(ts.Parents())
 		if err != nil {
 			return -1, xerrors.Errorf("getting parent tipset: %w", err)
 		}
@@ -180,7 +241,7 @@ func (a *GasAPI) GasEstimateGasLimit(ctx context.Context, msgIn *types.Message,
 	}
 
 	// Special case for PaymentChannel collect, which is deleting actor
-	st, err := a.Stmgr.ParentState(ts)
+	st, err := smgr.ParentState(ts)
 	if err != nil {
 		_ = err
 		// somewhat ignore it as it can happen and we just want to detect
@@ -206,17 +267,17 @@ func (a *GasAPI) GasEstimateGasLimit(ctx context.Context, msgIn *types.Message,
 	return res.MsgRct.GasUsed + 76e3, nil
 }
 
-func (a *GasAPI) GasEstimateMessageGas(ctx context.Context, msg *types.Message, spec *api.MessageSendSpec, _ types.TipSetKey) (*types.Message, error) {
+func (m *GasModule) GasEstimateMessageGas(ctx context.Context, msg *types.Message, spec *api.MessageSendSpec, _ types.TipSetKey) (*types.Message, error) {
 	if msg.GasLimit == 0 {
-		gasLimit, err := a.GasEstimateGasLimit(ctx, msg, types.TipSetKey{})
+		gasLimit, err := m.GasEstimateGasLimit(ctx, msg, types.TipSetKey{})
 		if err != nil {
 			return nil, xerrors.Errorf("estimating gas used: %w", err)
 		}
-		msg.GasLimit = int64(float64(gasLimit) * a.Mpool.GetConfig().GasLimitOverestimation)
+		msg.GasLimit = int64(float64(gasLimit) * m.Mpool.GetConfig().GasLimitOverestimation)
 	}
 
 	if msg.GasPremium == types.EmptyInt || types.BigCmp(msg.GasPremium, types.NewInt(0)) == 0 {
-		gasPremium, err := a.GasEstimateGasPremium(ctx, 2, msg.From, msg.GasLimit, types.TipSetKey{})
+		gasPremium, err := m.GasEstimateGasPremium(ctx, 2, msg.From, msg.GasLimit, types.TipSetKey{})
 		if err != nil {
 			return nil, xerrors.Errorf("estimating gas price: %w", err)
 		}
@@ -224,7 +285,7 @@ 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, 20, types.EmptyTSK)
+		feeCap, err := m.GasEstimateFeeCap(ctx, msg, 20, types.EmptyTSK)
 		if err != nil {
 			return nil, xerrors.Errorf("estimating fee cap: %w", err)
 		}

node/impl/full/mpool.go

@@ -10,14 +10,32 @@ import (
 	"golang.org/x/xerrors"
 
 	"github.com/filecoin-project/lotus/api"
+	"github.com/filecoin-project/lotus/chain/messagepool"
 	"github.com/filecoin-project/lotus/chain/messagesigner"
 	"github.com/filecoin-project/lotus/chain/types"
 	"github.com/filecoin-project/lotus/node/modules/dtypes"
 )
 
+type MpoolModuleAPI interface {
+	MpoolPush(ctx context.Context, smsg *types.SignedMessage) (cid.Cid, error)
+}
+
+// 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).
+type MpoolModule struct {
+	fx.In
+
+	Mpool *messagepool.MessagePool
+}
+
+var _ MpoolModuleAPI = (*MpoolModule)(nil)
+
 type MpoolAPI struct {
 	fx.In
 
+	MpoolModuleAPI
+
 	WalletAPI
 	GasAPI
 
@@ -106,8 +124,8 @@ func (a *MpoolAPI) MpoolClear(ctx context.Context, local bool) error {
 	return nil
 }
 
-func (a *MpoolAPI) MpoolPush(ctx context.Context, smsg *types.SignedMessage) (cid.Cid, error) {
-	return a.Mpool.Push(smsg)
+func (m *MpoolModule) MpoolPush(ctx context.Context, smsg *types.SignedMessage) (cid.Cid, error) {
+	return m.Mpool.Push(smsg)
 }
 
 func (a *MpoolAPI) MpoolPushUntrusted(ctx context.Context, smsg *types.SignedMessage) (cid.Cid, error) {
@@ -162,7 +180,7 @@ func (a *MpoolAPI) MpoolPushMessage(ctx context.Context, msg *types.Message, spe
 
 	// Sign and push the message
 	return a.MessageSigner.SignMessage(ctx, msg, func(smsg *types.SignedMessage) error {
-		if _, err := a.Mpool.Push(smsg); err != nil {
+		if _, err := a.MpoolModuleAPI.MpoolPush(ctx, smsg); err != nil {
 			return xerrors.Errorf("mpool push: failed to push message: %w", err)
 		}
 		return nil

node/impl/full/multisig.go

@@ -23,9 +23,8 @@ import (
 type MsigAPI struct {
 	fx.In
 
-	WalletAPI WalletAPI
-	StateAPI  StateAPI
-	MpoolAPI  MpoolAPI
+	StateAPI StateAPI
+	MpoolAPI MpoolAPI
 }
 
 func (a *MsigAPI) messageBuilder(ctx context.Context, from address.Address) (multisig.MessageBuilder, error) {
@@ -95,7 +94,7 @@ func (a *MsigAPI) MsigAddApprove(ctx context.Context, msig address.Address, src
 		return cid.Undef, actErr
 	}
 
-	return a.MsigApprove(ctx, msig, txID, proposer, msig, big.Zero(), src, uint64(builtin0.MethodsMultisig.AddSigner), enc)
+	return a.MsigApproveTxnHash(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) {
@@ -122,7 +121,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(builtin0.MethodsMultisig.SwapSigner), enc)
+	return a.MsigApproveTxnHash(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) {
@@ -134,15 +133,64 @@ func (a *MsigAPI) MsigSwapCancel(ctx context.Context, msig address.Address, src
 	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) {
-	return a.msigApproveOrCancel(ctx, api.MsigApprove, msig, txID, proposer, to, amt, src, method, params)
+func (a *MsigAPI) MsigApprove(ctx context.Context, msig address.Address, txID uint64, src address.Address) (cid.Cid, 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) {
+	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) {
-	return a.msigApproveOrCancel(ctx, api.MsigCancel, msig, txID, src, to, amt, src, method, params)
+	return a.msigApproveOrCancelTxnHash(ctx, api.MsigCancel, msig, txID, src, to, amt, src, method, params)
 }
 
-func (a *MsigAPI) msigApproveOrCancel(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) MsigRemoveSigner(ctx context.Context, msig address.Address, proposer address.Address, toRemove address.Address, decrease bool) (cid.Cid, error) {
+	enc, actErr := serializeRemoveParams(toRemove, decrease)
+	if actErr != nil {
+		return cid.Undef, actErr
+	}
+
+	return a.MsigPropose(ctx, msig, msig, types.NewInt(0), proposer, uint64(builtin0.MethodsMultisig.RemoveSigner), enc)
+}
+
+func (a *MsigAPI) msigApproveOrCancelSimple(ctx context.Context, operation api.MsigProposeResponse, msig address.Address, txID uint64, src address.Address) (cid.Cid, error) {
+	if msig == address.Undef {
+		return cid.Undef, xerrors.Errorf("must provide multisig address")
+	}
+
+	if src == address.Undef {
+		return cid.Undef, xerrors.Errorf("must provide source address")
+	}
+
+	mb, err := a.messageBuilder(ctx, src)
+	if err != nil {
+		return cid.Undef, err
+	}
+
+	var msg *types.Message
+	switch operation {
+	case api.MsigApprove:
+		msg, err = mb.Approve(msig, txID, nil)
+	case api.MsigCancel:
+		msg, err = mb.Cancel(msig, txID, nil)
+	default:
+		return cid.Undef, xerrors.Errorf("Invalid operation for msigApproveOrCancel")
+	}
+	if err != nil {
+		return cid.Undef, err
+	}
+
+	smsg, err := a.MpoolAPI.MpoolPushMessage(ctx, msg, nil)
+	if err != nil {
+		return cid.Undef, err
+	}
+
+	return smsg.Cid(), 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) {
 	if msig == address.Undef {
 		return cid.Undef, xerrors.Errorf("must provide multisig address")
 	}
@@ -216,3 +264,15 @@ func serializeSwapParams(old address.Address, new address.Address) ([]byte, erro
 
 	return enc, nil
 }
+
+func serializeRemoveParams(rem address.Address, dec bool) ([]byte, error) {
+	enc, actErr := actors.SerializeParams(&multisig0.RemoveSignerParams{
+		Signer:   rem,
+		Decrease: dec,
+	})
+	if actErr != nil {
+		return nil, actErr
+	}
+
+	return enc, nil
+}

node/impl/full/state.go

@@ -5,14 +5,6 @@ import (
 	"context"
 	"strconv"
 
-	"github.com/filecoin-project/lotus/chain/actors/builtin"
-	"github.com/filecoin-project/lotus/chain/actors/policy"
-
-	"github.com/filecoin-project/lotus/chain/actors/builtin/verifreg"
-
-	"github.com/filecoin-project/go-state-types/dline"
-	"github.com/filecoin-project/go-state-types/network"
-
 	cid "github.com/ipfs/go-cid"
 	cbor "github.com/ipfs/go-ipld-cbor"
 	"go.uber.org/fx"
@@ -22,14 +14,19 @@ import (
 	"github.com/filecoin-project/go-bitfield"
 	"github.com/filecoin-project/go-state-types/abi"
 	"github.com/filecoin-project/go-state-types/big"
+	"github.com/filecoin-project/go-state-types/dline"
+	"github.com/filecoin-project/go-state-types/network"
 	"github.com/filecoin-project/lotus/extern/sector-storage/ffiwrapper"
 
 	"github.com/filecoin-project/lotus/api"
+	"github.com/filecoin-project/lotus/chain/actors/builtin"
 	"github.com/filecoin-project/lotus/chain/actors/builtin/market"
 	"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
 	"github.com/filecoin-project/lotus/chain/actors/builtin/multisig"
 	"github.com/filecoin-project/lotus/chain/actors/builtin/power"
 	"github.com/filecoin-project/lotus/chain/actors/builtin/reward"
+	"github.com/filecoin-project/lotus/chain/actors/builtin/verifreg"
+	"github.com/filecoin-project/lotus/chain/actors/policy"
 	"github.com/filecoin-project/lotus/chain/beacon"
 	"github.com/filecoin-project/lotus/chain/gen"
 	"github.com/filecoin-project/lotus/chain/state"
@@ -42,12 +39,36 @@ import (
 	"github.com/filecoin-project/lotus/node/modules/dtypes"
 )
 
+type StateModuleAPI interface {
+	StateAccountKey(ctx context.Context, addr address.Address, tsk types.TipSetKey) (address.Address, error)
+	StateGetActor(ctx context.Context, actor address.Address, tsk types.TipSetKey) (*types.Actor, error)
+	StateLookupID(ctx context.Context, addr address.Address, tsk types.TipSetKey) (address.Address, error)
+	StateWaitMsg(ctx context.Context, msg cid.Cid, confidence uint64) (*api.MsgLookup, error)
+	MsigGetAvailableBalance(ctx context.Context, addr address.Address, tsk types.TipSetKey) (types.BigInt, error)
+	MsigGetVested(ctx context.Context, addr address.Address, start types.TipSetKey, end types.TipSetKey) (types.BigInt, error)
+}
+
+// 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).
+type StateModule struct {
+	fx.In
+
+	StateManager *stmgr.StateManager
+	Chain        *store.ChainStore
+}
+
+var _ StateModuleAPI = (*StateModule)(nil)
+
 type StateAPI struct {
 	fx.In
 
 	// TODO: the wallet here is only needed because we have the MinerCreateBlock
 	// API attached to the state API. It probably should live somewhere better
-	Wallet *wallet.Wallet
+	Wallet    api.WalletAPI
+	DefWallet wallet.Default
+
+	StateModuleAPI
 
 	ProofVerifier ffiwrapper.Verifier
 	StateManager  *stmgr.StateManager
@@ -349,27 +370,33 @@ func (a *StateAPI) StateReplay(ctx context.Context, tsk types.TipSetKey, mc cid.
 	}, nil
 }
 
-func (a *StateAPI) stateForTs(ctx context.Context, ts *types.TipSet) (*state.StateTree, error) {
+func stateForTs(ctx context.Context, ts *types.TipSet, cstore *store.ChainStore, smgr *stmgr.StateManager) (*state.StateTree, error) {
 	if ts == nil {
-		ts = a.Chain.GetHeaviestTipSet()
+		ts = cstore.GetHeaviestTipSet()
 	}
 
-	st, _, err := a.StateManager.TipSetState(ctx, ts)
+	st, _, err := smgr.TipSetState(ctx, ts)
 	if err != nil {
 		return nil, err
 	}
 
-	buf := bufbstore.NewBufferedBstore(a.Chain.Blockstore())
+	buf := bufbstore.NewBufferedBstore(cstore.Blockstore())
 	cst := cbor.NewCborStore(buf)
 	return state.LoadStateTree(cst, st)
 }
+func (a *StateAPI) stateForTs(ctx context.Context, ts *types.TipSet) (*state.StateTree, error) {
+	return stateForTs(ctx, ts, a.Chain, a.StateManager)
+}
+func (m *StateModule) stateForTs(ctx context.Context, ts *types.TipSet) (*state.StateTree, error) {
+	return stateForTs(ctx, ts, m.Chain, m.StateManager)
+}
 
-func (a *StateAPI) StateGetActor(ctx context.Context, actor address.Address, tsk types.TipSetKey) (*types.Actor, error) {
-	ts, err := a.Chain.GetTipSetFromKey(tsk)
+func (m *StateModule) StateGetActor(ctx context.Context, actor address.Address, tsk types.TipSetKey) (*types.Actor, error) {
+	ts, err := m.Chain.GetTipSetFromKey(tsk)
 	if err != nil {
 		return nil, xerrors.Errorf("loading tipset %s: %w", tsk, err)
 	}
-	state, err := a.stateForTs(ctx, ts)
+	state, err := m.stateForTs(ctx, ts)
 	if err != nil {
 		return nil, xerrors.Errorf("computing tipset state failed: %w", err)
 	}
@@ -377,26 +404,22 @@ func (a *StateAPI) StateGetActor(ctx context.Context, actor address.Address, tsk
 	return state.GetActor(actor)
 }
 
-func (a *StateAPI) StateLookupID(ctx context.Context, addr address.Address, tsk types.TipSetKey) (address.Address, error) {
-	ts, err := a.Chain.GetTipSetFromKey(tsk)
+func (m *StateModule) StateLookupID(ctx context.Context, addr address.Address, tsk types.TipSetKey) (address.Address, error) {
+	ts, err := m.Chain.GetTipSetFromKey(tsk)
 	if err != nil {
 		return address.Undef, xerrors.Errorf("loading tipset %s: %w", tsk, err)
 	}
-	state, err := a.stateForTs(ctx, ts)
-	if err != nil {
-		return address.Undef, err
-	}
 
-	return state.LookupID(addr)
+	return m.StateManager.LookupID(ctx, addr, ts)
 }
 
-func (a *StateAPI) StateAccountKey(ctx context.Context, addr address.Address, tsk types.TipSetKey) (address.Address, error) {
-	ts, err := a.Chain.GetTipSetFromKey(tsk)
+func (m *StateModule) StateAccountKey(ctx context.Context, addr address.Address, tsk types.TipSetKey) (address.Address, error) {
+	ts, err := m.Chain.GetTipSetFromKey(tsk)
 	if err != nil {
 		return address.Undef, xerrors.Errorf("loading tipset %s: %w", tsk, err)
 	}
 
-	return a.StateManager.ResolveToKeyAddress(ctx, addr, ts)
+	return m.StateManager.ResolveToKeyAddress(ctx, addr, ts)
 }
 
 func (a *StateAPI) StateReadState(ctx context.Context, actor address.Address, tsk types.TipSetKey) (*api.ActorState, error) {
@@ -453,22 +476,28 @@ func (a *StateAPI) MinerCreateBlock(ctx context.Context, bt *api.BlockTemplate)
 	return &out, nil
 }
 
-func (a *StateAPI) StateWaitMsg(ctx context.Context, msg cid.Cid, confidence uint64) (*api.MsgLookup, error) {
-	ts, recpt, found, err := a.StateManager.WaitForMessage(ctx, msg, confidence)
+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)
 	if err != nil {
 		return nil, err
 	}
 
 	var returndec interface{}
 	if recpt.ExitCode == 0 && len(recpt.Return) > 0 {
-		cmsg, err := a.Chain.GetCMessage(msg)
+		cmsg, err := cstore.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, a.StateManager, vmsg.To, vmsg.Method, ts)
+		t, err := stmgr.GetReturnType(ctx, smgr, vmsg.To, vmsg.Method, ts)
 		if err != nil {
 			return nil, xerrors.Errorf("failed to get return type: %w", err)
 		}
@@ -799,17 +828,17 @@ func (a *StateAPI) StateCompute(ctx context.Context, height abi.ChainEpoch, msgs
 	}, nil
 }
 
-func (a *StateAPI) MsigGetAvailableBalance(ctx context.Context, addr address.Address, tsk types.TipSetKey) (types.BigInt, error) {
-	ts, err := a.Chain.GetTipSetFromKey(tsk)
+func (m *StateModule) MsigGetAvailableBalance(ctx context.Context, addr address.Address, tsk types.TipSetKey) (types.BigInt, error) {
+	ts, err := m.Chain.GetTipSetFromKey(tsk)
 	if err != nil {
 		return types.EmptyInt, xerrors.Errorf("loading tipset %s: %w", tsk, err)
 	}
 
-	act, err := a.StateManager.LoadActor(ctx, addr, ts)
+	act, err := m.StateManager.LoadActor(ctx, addr, ts)
 	if err != nil {
 		return types.EmptyInt, xerrors.Errorf("failed to load multisig actor: %w", err)
 	}
-	msas, err := multisig.Load(a.Chain.Store(ctx), act)
+	msas, err := multisig.Load(m.Chain.Store(ctx), act)
 	if err != nil {
 		return types.EmptyInt, xerrors.Errorf("failed to load multisig actor state: %w", err)
 	}
@@ -820,13 +849,51 @@ func (a *StateAPI) MsigGetAvailableBalance(ctx context.Context, addr address.Add
 	return types.BigSub(act.Balance, locked), nil
 }
 
-func (a *StateAPI) MsigGetVested(ctx context.Context, addr address.Address, start types.TipSetKey, end types.TipSetKey) (types.BigInt, error) {
-	startTs, err := a.Chain.GetTipSetFromKey(start)
+func (a *StateAPI) MsigGetVestingSchedule(ctx context.Context, addr address.Address, tsk types.TipSetKey) (api.MsigVesting, error) {
+	ts, err := a.Chain.GetTipSetFromKey(tsk)
+	if err != nil {
+		return api.EmptyVesting, xerrors.Errorf("loading tipset %s: %w", tsk, err)
+	}
+
+	act, err := a.StateManager.LoadActor(ctx, addr, ts)
+	if err != nil {
+		return api.EmptyVesting, xerrors.Errorf("failed to load multisig actor: %w", err)
+	}
+
+	msas, err := multisig.Load(a.Chain.Store(ctx), act)
+	if err != nil {
+		return api.EmptyVesting, xerrors.Errorf("failed to load multisig actor state: %w", err)
+	}
+
+	ib, err := msas.InitialBalance()
+	if err != nil {
+		return api.EmptyVesting, xerrors.Errorf("failed to load multisig initial balance: %w", err)
+	}
+
+	se, err := msas.StartEpoch()
+	if err != nil {
+		return api.EmptyVesting, xerrors.Errorf("failed to load multisig start epoch: %w", err)
+	}
+
+	ud, err := msas.UnlockDuration()
+	if err != nil {
+		return api.EmptyVesting, xerrors.Errorf("failed to load multisig unlock duration: %w", err)
+	}
+
+	return api.MsigVesting{
+		InitialBalance: ib,
+		StartEpoch:     se,
+		UnlockDuration: ud,
+	}, nil
+}
+
+func (m *StateModule) MsigGetVested(ctx context.Context, addr address.Address, start types.TipSetKey, end types.TipSetKey) (types.BigInt, error) {
+	startTs, err := m.Chain.GetTipSetFromKey(start)
 	if err != nil {
 		return types.EmptyInt, xerrors.Errorf("loading start tipset %s: %w", start, err)
 	}
 
-	endTs, err := a.Chain.GetTipSetFromKey(end)
+	endTs, err := m.Chain.GetTipSetFromKey(end)
 	if err != nil {
 		return types.EmptyInt, xerrors.Errorf("loading end tipset %s: %w", end, err)
 	}
@@ -837,12 +904,12 @@ func (a *StateAPI) MsigGetVested(ctx context.Context, addr address.Address, star
 		return big.Zero(), nil
 	}
 
-	act, err := a.StateManager.LoadActor(ctx, addr, endTs)
+	act, err := m.StateManager.LoadActor(ctx, addr, endTs)
 	if err != nil {
 		return types.EmptyInt, xerrors.Errorf("failed to load multisig actor at end epoch: %w", err)
 	}
 
-	msas, err := multisig.Load(a.Chain.Store(ctx), act)
+	msas, err := multisig.Load(m.Chain.Store(ctx), act)
 	if err != nil {
 		return types.EmptyInt, xerrors.Errorf("failed to load multisig actor state: %w", err)
 	}
@@ -982,7 +1049,7 @@ func (a *StateAPI) StateMinerInitialPledgeCollateral(ctx context.Context, maddr
 		return types.EmptyInt, xerrors.Errorf("loading reward actor state: %w", err)
 	}
 
-	circSupply, err := a.StateCirculatingSupply(ctx, ts.Key())
+	circSupply, err := a.StateVMCirculatingSupplyInternal(ctx, ts.Key())
 	if err != nil {
 		return big.Zero(), xerrors.Errorf("getting circulating supply: %w", err)
 	}
@@ -1136,7 +1203,7 @@ func (a *StateAPI) StateDealProviderCollateralBounds(ctx context.Context, size a
 		return api.DealCollateralBounds{}, xerrors.Errorf("failed to load reward actor state: %w", err)
 	}
 
-	circ, err := a.StateCirculatingSupply(ctx, ts.Key())
+	circ, err := a.StateVMCirculatingSupplyInternal(ctx, ts.Key())
 	if err != nil {
 		return api.DealCollateralBounds{}, xerrors.Errorf("getting total circulating supply: %w", err)
 	}
@@ -1164,7 +1231,20 @@ func (a *StateAPI) StateDealProviderCollateralBounds(ctx context.Context, size a
 	}, nil
 }
 
-func (a *StateAPI) StateCirculatingSupply(ctx context.Context, tsk types.TipSetKey) (api.CirculatingSupply, error) {
+func (a *StateAPI) StateCirculatingSupply(ctx context.Context, tsk types.TipSetKey) (abi.TokenAmount, error) {
+	ts, err := a.Chain.GetTipSetFromKey(tsk)
+	if err != nil {
+		return types.EmptyInt, xerrors.Errorf("loading tipset %s: %w", tsk, err)
+	}
+
+	sTree, err := a.stateForTs(ctx, ts)
+	if err != nil {
+		return types.EmptyInt, err
+	}
+	return a.StateManager.GetCirculatingSupply(ctx, ts.Height(), sTree)
+}
+
+func (a *StateAPI) StateVMCirculatingSupplyInternal(ctx context.Context, tsk types.TipSetKey) (api.CirculatingSupply, error) {
 	ts, err := a.Chain.GetTipSetFromKey(tsk)
 	if err != nil {
 		return api.CirculatingSupply{}, xerrors.Errorf("loading tipset %s: %w", tsk, err)
@@ -1174,7 +1254,7 @@ func (a *StateAPI) StateCirculatingSupply(ctx context.Context, tsk types.TipSetK
 	if err != nil {
 		return api.CirculatingSupply{}, err
 	}
-	return a.StateManager.GetCirculatingSupplyDetailed(ctx, ts.Height(), sTree)
+	return a.StateManager.GetVMCirculatingSupplyDetailed(ctx, ts.Height(), sTree)
 }
 
 func (a *StateAPI) StateNetworkVersion(ctx context.Context, tsk types.TipSetKey) (network.Version, error) {

node/impl/full/sync.go

@@ -118,6 +118,12 @@ func (a *SyncAPI) SyncUnmarkBad(ctx context.Context, bcid cid.Cid) error {
 	return nil
 }
 
+func (a *SyncAPI) SyncUnmarkAllBad(ctx context.Context) error {
+	log.Warnf("Dropping bad block cache")
+	a.Syncer.UnmarkAllBad()
+	return nil
+}
+
 func (a *SyncAPI) SyncCheckBad(ctx context.Context, bcid cid.Cid) (string, error) {
 	reason, ok := a.Syncer.CheckBadBlockCache(bcid)
 	if !ok {

node/impl/full/wallet.go

@@ -3,13 +3,14 @@ package full
 import (
 	"context"
 
-	"github.com/filecoin-project/go-address"
 	"go.uber.org/fx"
 	"golang.org/x/xerrors"
 
+	"github.com/filecoin-project/go-address"
 	"github.com/filecoin-project/go-state-types/big"
 	"github.com/filecoin-project/go-state-types/crypto"
 
+	"github.com/filecoin-project/lotus/api"
 	"github.com/filecoin-project/lotus/chain/stmgr"
 	"github.com/filecoin-project/lotus/chain/types"
 	"github.com/filecoin-project/lotus/chain/wallet"
@@ -19,24 +20,13 @@ import (
 type WalletAPI struct {
 	fx.In
 
-	StateManager *stmgr.StateManager
-	Wallet       *wallet.Wallet
-}
-
-func (a *WalletAPI) WalletNew(ctx context.Context, typ crypto.SigType) (address.Address, error) {
-	return a.Wallet.GenerateKey(typ)
-}
-
-func (a *WalletAPI) WalletHas(ctx context.Context, addr address.Address) (bool, error) {
-	return a.Wallet.HasKey(addr)
-}
-
-func (a *WalletAPI) WalletList(ctx context.Context) ([]address.Address, error) {
-	return a.Wallet.ListAddrs()
+	StateManagerAPI stmgr.StateManagerAPI
+	Default         wallet.Default
+	api.WalletAPI
 }
 
 func (a *WalletAPI) WalletBalance(ctx context.Context, addr address.Address) (types.BigInt, error) {
-	act, err := a.StateManager.LoadActorTsk(ctx, addr, types.EmptyTSK)
+	act, err := a.StateManagerAPI.LoadActorTsk(ctx, addr, types.EmptyTSK)
 	if xerrors.Is(err, types.ErrActorNotFound) {
 		return big.Zero(), nil
 	} else if err != nil {
@@ -46,17 +36,30 @@ func (a *WalletAPI) WalletBalance(ctx context.Context, addr address.Address) (ty
 }
 
 func (a *WalletAPI) WalletSign(ctx context.Context, k address.Address, msg []byte) (*crypto.Signature, error) {
-	keyAddr, err := a.StateManager.ResolveToKeyAddress(ctx, k, nil)
+	keyAddr, err := a.StateManagerAPI.ResolveToKeyAddress(ctx, k, nil)
 	if err != nil {
 		return nil, xerrors.Errorf("failed to resolve ID address: %w", keyAddr)
 	}
-	return a.Wallet.Sign(ctx, keyAddr, msg)
+	return a.WalletAPI.WalletSign(ctx, keyAddr, msg, api.MsgMeta{
+		Type: api.MTUnknown,
+	})
 }
 
 func (a *WalletAPI) WalletSignMessage(ctx context.Context, k address.Address, msg *types.Message) (*types.SignedMessage, error) {
-	mcid := msg.Cid()
+	keyAddr, err := a.StateManagerAPI.ResolveToKeyAddress(ctx, k, nil)
+	if err != nil {
+		return nil, xerrors.Errorf("failed to resolve ID address: %w", keyAddr)
+	}
 
-	sig, err := a.WalletSign(ctx, k, mcid.Bytes())
+	mb, err := msg.ToStorageBlock()
+	if err != nil {
+		return nil, xerrors.Errorf("serializing message: %w", err)
+	}
+
+	sig, err := a.WalletAPI.WalletSign(ctx, k, mb.Cid().Bytes(), api.MsgMeta{
+		Type:  api.MTChainMsg,
+		Extra: mb.RawData(),
+	})
 	if err != nil {
 		return nil, xerrors.Errorf("failed to sign message: %w", err)
 	}
@@ -72,23 +75,11 @@ func (a *WalletAPI) WalletVerify(ctx context.Context, k address.Address, msg []b
 }
 
 func (a *WalletAPI) WalletDefaultAddress(ctx context.Context) (address.Address, error) {
-	return a.Wallet.GetDefault()
+	return a.Default.GetDefault()
 }
 
 func (a *WalletAPI) WalletSetDefault(ctx context.Context, addr address.Address) error {
-	return a.Wallet.SetDefault(addr)
-}
-
-func (a *WalletAPI) WalletExport(ctx context.Context, addr address.Address) (*types.KeyInfo, error) {
-	return a.Wallet.Export(addr)
-}
-
-func (a *WalletAPI) WalletImport(ctx context.Context, ki *types.KeyInfo) (address.Address, error) {
-	return a.Wallet.Import(ki)
-}
-
-func (a *WalletAPI) WalletDelete(ctx context.Context, addr address.Address) error {
-	return a.Wallet.DeleteKey(addr)
+	return a.Default.SetDefault(addr)
 }
 
 func (a *WalletAPI) WalletValidateAddress(ctx context.Context, str string) (address.Address, error) {

node/modules/chain.go

@@ -17,6 +17,7 @@ import (
 	"golang.org/x/xerrors"
 
 	"github.com/filecoin-project/lotus/extern/sector-storage/ffiwrapper"
+	"github.com/filecoin-project/lotus/journal"
 
 	"github.com/filecoin-project/lotus/build"
 	"github.com/filecoin-project/lotus/chain"
@@ -51,9 +52,9 @@ func ChainBitswap(mctx helpers.MetricsCtx, lc fx.Lifecycle, host host.Host, rt r
 	return exch
 }
 
-func MessagePool(lc fx.Lifecycle, sm *stmgr.StateManager, ps *pubsub.PubSub, ds dtypes.MetadataDS, nn dtypes.NetworkName) (*messagepool.MessagePool, error) {
+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)
-	mp, err := messagepool.New(mpp, ds, nn)
+	mp, err := messagepool.New(mpp, ds, nn, j)
 	if err != nil {
 		return nil, xerrors.Errorf("constructing mpool: %w", err)
 	}
@@ -88,8 +89,8 @@ func ChainBlockService(bs dtypes.ChainBlockstore, rem dtypes.ChainBitswap) dtype
 	return blockservice.New(bs, rem)
 }
 
-func ChainStore(lc fx.Lifecycle, bs dtypes.ChainBlockstore, ds dtypes.MetadataDS, syscalls vm.SyscallBuilder) *store.ChainStore {
-	chain := store.NewChainStore(bs, ds, syscalls)
+func ChainStore(lc fx.Lifecycle, bs dtypes.ChainBlockstore, ds dtypes.MetadataDS, syscalls vm.SyscallBuilder, j journal.Journal) *store.ChainStore {
+	chain := store.NewChainStore(bs, ds, syscalls, j)
 
 	if err := chain.Load(); err != nil {
 		log.Warnf("loading chain state from disk: %s", err)

node/modules/client.go

@@ -113,7 +113,7 @@ 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 *discoveryimpl.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, j journal.Journal) (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 {
@@ -124,8 +124,8 @@ func StorageClient(lc fx.Lifecycle, h host.Host, ibs dtypes.ClientBlockstore, md
 		OnStart: func(ctx context.Context) error {
 			c.SubscribeToEvents(marketevents.StorageClientLogger)
 
-			evtType := journal.J.RegisterEventType("markets/storage/client", "state_change")
-			c.SubscribeToEvents(markets.StorageClientJournaler(evtType))
+			evtType := j.RegisterEventType("markets/storage/client", "state_change")
+			c.SubscribeToEvents(markets.StorageClientJournaler(j, evtType))
 
 			return c.Start(ctx)
 		},
@@ -137,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 discovery.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, j journal.Journal) (retrievalmarket.RetrievalClient, error) {
 	adapter := retrievaladapter.NewRetrievalClientNode(payAPI, chainAPI, stateAPI)
 	network := rmnet.NewFromLibp2pHost(h)
 	sc := storedcounter.New(ds, datastore.NewKey("/retr"))
@@ -150,8 +150,8 @@ func RetrievalClient(lc fx.Lifecycle, h host.Host, mds dtypes.ClientMultiDstore,
 		OnStart: func(ctx context.Context) error {
 			client.SubscribeToEvents(marketevents.RetrievalClientLogger)
 
-			evtType := journal.J.RegisterEventType("markets/retrieval/client", "state_change")
-			client.SubscribeToEvents(markets.RetrievalClientJournaler(evtType))
+			evtType := j.RegisterEventType("markets/retrieval/client", "state_change")
+			client.SubscribeToEvents(markets.RetrievalClientJournaler(j, evtType))
 
 			return client.Start(ctx)
 		},

node/modules/lp2p/libp2p.go

@@ -20,8 +20,8 @@ import (
 var log = logging.Logger("p2pnode")
 
 const (
-	KLibp2pHost  = "libp2p-host"
-	KTLibp2pHost = KLibp2pHost
+	KLibp2pHost                = "libp2p-host"
+	KTLibp2pHost types.KeyType = KLibp2pHost
 )
 
 type Libp2pOpts struct {

node/modules/lp2p/pubsub.go

@@ -187,6 +187,7 @@ func GossipSub(in GossipIn) (service *pubsub.PubSub, err error) {
 		build.MessagesTopic(in.Nn): 1,
 	}
 
+	var drandTopics []string
 	for _, d := range in.Dr {
 		topic, err := getDrandTopic(d.Config.ChainInfoJSON)
 		if err != nil {
@@ -194,6 +195,7 @@ func GossipSub(in GossipIn) (service *pubsub.PubSub, err error) {
 		}
 		topicParams[topic] = drandTopicParams
 		pgTopicWeights[topic] = 5
+		drandTopics = append(drandTopics, topic)
 	}
 
 	options := []pubsub.Option{
@@ -309,6 +311,17 @@ func GossipSub(in GossipIn) (service *pubsub.PubSub, err error) {
 
 	options = append(options, pubsub.WithPeerGater(pgParams))
 
+	allowTopics := []string{
+		build.BlocksTopic(in.Nn),
+		build.MessagesTopic(in.Nn),
+	}
+	allowTopics = append(allowTopics, drandTopics...)
+	options = append(options,
+		pubsub.WithSubscriptionFilter(
+			pubsub.WrapLimitSubscriptionFilter(
+				pubsub.NewAllowlistSubscriptionFilter(allowTopics...),
+				100)))
+
 	// tracer
 	if in.Cfg.RemoteTracer != "" {
 		a, err := ma.NewMultiaddr(in.Cfg.RemoteTracer)
@@ -359,14 +372,9 @@ type tracerWrapper struct {
 	topics map[string]struct{}
 }
 
-func (trw *tracerWrapper) traceMessage(topics []string) bool {
-	for _, topic := range topics {
-		_, ok := trw.topics[topic]
-		if ok {
-			return true
-		}
-	}
-	return false
+func (trw *tracerWrapper) traceMessage(topic string) bool {
+	_, ok := trw.topics[topic]
+	return ok
 }
 
 func (trw *tracerWrapper) Trace(evt *pubsub_pb.TraceEvent) {
@@ -379,12 +387,12 @@ func (trw *tracerWrapper) Trace(evt *pubsub_pb.TraceEvent) {
 	switch evt.GetType() {
 	case pubsub_pb.TraceEvent_PUBLISH_MESSAGE:
 		stats.Record(context.TODO(), metrics.PubsubPublishMessage.M(1))
-		if trw.tr != nil && trw.traceMessage(evt.GetPublishMessage().Topics) {
+		if trw.tr != nil && trw.traceMessage(evt.GetPublishMessage().GetTopic()) {
 			trw.tr.Trace(evt)
 		}
 	case pubsub_pb.TraceEvent_DELIVER_MESSAGE:
 		stats.Record(context.TODO(), metrics.PubsubDeliverMessage.M(1))
-		if trw.tr != nil && trw.traceMessage(evt.GetDeliverMessage().Topics) {
+		if trw.tr != nil && trw.traceMessage(evt.GetDeliverMessage().GetTopic()) {
 			trw.tr.Trace(evt)
 		}
 	case pubsub_pb.TraceEvent_REJECT_MESSAGE:

node/modules/mpoolnonceapi.go

@@ -0,0 +1,33 @@
+package modules
+
+import (
+	"context"
+
+	"go.uber.org/fx"
+
+	"github.com/filecoin-project/lotus/node/impl/full"
+
+	"github.com/filecoin-project/lotus/chain/messagesigner"
+	"github.com/filecoin-project/lotus/chain/types"
+
+	"github.com/filecoin-project/go-address"
+)
+
+// MpoolNonceAPI substitutes the mpool nonce with an implementation that
+// doesn't rely on the mpool - it just gets the nonce from actor state
+type MpoolNonceAPI struct {
+	fx.In
+
+	StateAPI full.StateAPI
+}
+
+// GetNonce gets the nonce from actor state
+func (a *MpoolNonceAPI) GetNonce(addr address.Address) (uint64, error) {
+	act, err := a.StateAPI.StateGetActor(context.Background(), addr, types.EmptyTSK)
+	if err != nil {
+		return 0, err
+	}
+	return act.Nonce, nil
+}
+
+var _ messagesigner.MpoolNonceAPI = (*MpoolNonceAPI)(nil)

node/modules/rpcstatemanager.go

@@ -0,0 +1,33 @@
+package modules
+
+import (
+	"context"
+
+	"github.com/filecoin-project/lotus/api"
+
+	"github.com/filecoin-project/go-address"
+	"github.com/filecoin-project/lotus/chain/stmgr"
+	"github.com/filecoin-project/lotus/chain/types"
+)
+
+type RPCStateManager struct {
+	gapi api.GatewayAPI
+}
+
+func NewRPCStateManager(api api.GatewayAPI) *RPCStateManager {
+	return &RPCStateManager{gapi: api}
+}
+
+func (s *RPCStateManager) LoadActorTsk(ctx context.Context, addr address.Address, tsk types.TipSetKey) (*types.Actor, error) {
+	return s.gapi.StateGetActor(ctx, addr, tsk)
+}
+
+func (s *RPCStateManager) LookupID(ctx context.Context, addr address.Address, ts *types.TipSet) (address.Address, error) {
+	return s.gapi.StateLookupID(ctx, addr, ts.Key())
+}
+
+func (s *RPCStateManager) ResolveToKeyAddress(ctx context.Context, addr address.Address, ts *types.TipSet) (address.Address, error) {
+	return s.gapi.StateAccountKey(ctx, addr, ts.Key())
+}
+
+var _ stmgr.StateManagerAPI = (*RPCStateManager)(nil)

node/modules/services.go

@@ -42,11 +42,13 @@ func RunHello(mctx helpers.MetricsCtx, lc fx.Lifecycle, h host.Host, svc *hello.
 		return xerrors.Errorf("failed to subscribe to event bus: %w", err)
 	}
 
+	ctx := helpers.LifecycleCtx(mctx, lc)
+
 	go func() {
 		for evt := range sub.Out() {
 			pic := evt.(event.EvtPeerIdentificationCompleted)
 			go func() {
-				if err := svc.SayHello(helpers.LifecycleCtx(mctx, lc), pic.Peer); err != nil {
+				if err := svc.SayHello(ctx, pic.Peer); err != nil {
 					protos, _ := h.Peerstore().GetProtocols(pic.Peer)
 					agent, _ := h.Peerstore().Get(pic.Peer, "AgentVersion")
 					if protosContains(protos, hello.ProtocolID) {

node/modules/storageminer.go

@@ -162,6 +162,7 @@ type StorageMinerParams struct {
 	SectorIDCounter    sealing.SectorIDCounter
 	Verifier           ffiwrapper.Verifier
 	GetSealingConfigFn dtypes.GetSealingConfigFunc
+	Journal            journal.Journal
 }
 
 func StorageMiner(fc config.MinerFeeConfig) func(params StorageMinerParams) (*storage.Miner, error) {
@@ -176,6 +177,7 @@ func StorageMiner(fc config.MinerFeeConfig) func(params StorageMinerParams) (*st
 			sc     = params.SectorIDCounter
 			verif  = params.Verifier
 			gsd    = params.GetSealingConfigFn
+			j      = params.Journal
 		)
 
 		maddr, err := minerAddrFromDS(ds)
@@ -195,12 +197,12 @@ func StorageMiner(fc config.MinerFeeConfig) func(params StorageMinerParams) (*st
 			return nil, err
 		}
 
-		fps, err := storage.NewWindowedPoStScheduler(api, fc, sealer, sealer, maddr, worker)
+		fps, err := storage.NewWindowedPoStScheduler(api, fc, sealer, sealer, j, maddr, worker)
 		if err != nil {
 			return nil, err
 		}
 
-		sm, err := storage.NewMiner(api, maddr, worker, h, ds, sealer, sc, verif, gsd, fc)
+		sm, err := storage.NewMiner(api, maddr, worker, h, ds, sealer, sc, verif, gsd, fc, j)
 		if err != nil {
 			return nil, err
 		}
@@ -217,15 +219,15 @@ func StorageMiner(fc config.MinerFeeConfig) func(params StorageMinerParams) (*st
 	}
 }
 
-func HandleRetrieval(host host.Host, lc fx.Lifecycle, m retrievalmarket.RetrievalProvider) {
+func HandleRetrieval(host host.Host, lc fx.Lifecycle, m retrievalmarket.RetrievalProvider, j journal.Journal) {
 	m.OnReady(marketevents.ReadyLogger("retrieval provider"))
 	lc.Append(fx.Hook{
 
 		OnStart: func(ctx context.Context) error {
 			m.SubscribeToEvents(marketevents.RetrievalProviderLogger)
 
-			evtType := journal.J.RegisterEventType("markets/retrieval/provider", "state_change")
-			m.SubscribeToEvents(markets.RetrievalProviderJournaler(evtType))
+			evtType := j.RegisterEventType("markets/retrieval/provider", "state_change")
+			m.SubscribeToEvents(markets.RetrievalProviderJournaler(j, evtType))
 
 			return m.Start(ctx)
 		},
@@ -235,15 +237,15 @@ 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) {
+func HandleDeals(mctx helpers.MetricsCtx, lc fx.Lifecycle, host host.Host, h storagemarket.StorageProvider, j journal.Journal) {
 	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))
+			evtType := j.RegisterEventType("markets/storage/provider", "state_change")
+			h.SubscribeToEvents(markets.StorageProviderJournaler(j, evtType))
 
 			return h.Start(ctx)
 		},
@@ -355,13 +357,13 @@ 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) (*miner.Miner, error) {
+func SetupBlockProducer(lc fx.Lifecycle, ds dtypes.MetadataDS, api lapi.FullNode, epp gen.WinningPoStProver, sf *slashfilter.SlashFilter, j journal.Journal) (*miner.Miner, error) {
 	minerAddr, err := minerAddrFromDS(ds)
 	if err != nil {
 		return nil, err
 	}
 
-	m := miner.NewMiner(api, epp, minerAddr, sf)
+	m := miner.NewMiner(api, epp, minerAddr, sf, j)
 
 	lc.Append(fx.Hook{
 		OnStart: func(ctx context.Context) error {
@@ -472,7 +474,7 @@ func BasicDealFilter(user dtypes.DealFilter) func(onlineOk dtypes.ConsiderOnline
 
 			// Reject if it's more than 7 days in the future
 			// TODO: read from cfg
-			maxStartEpoch := ht + abi.ChainEpoch(7*builtin.EpochsInDay)
+			maxStartEpoch := earliest + abi.ChainEpoch(7*builtin.EpochsInDay)
 			if deal.Proposal.StartEpoch > maxStartEpoch {
 				return false, fmt.Sprintf("deal start epoch is too far in the future: %s > %s", deal.Proposal.StartEpoch, maxStartEpoch), nil
 			}

node/modules/testing/genesis.go

@@ -23,17 +23,18 @@ import (
 	"github.com/filecoin-project/lotus/chain/types"
 	"github.com/filecoin-project/lotus/chain/vm"
 	"github.com/filecoin-project/lotus/genesis"
+	"github.com/filecoin-project/lotus/journal"
 	"github.com/filecoin-project/lotus/node/modules"
 	"github.com/filecoin-project/lotus/node/modules/dtypes"
 )
 
 var glog = logging.Logger("genesis")
 
-func MakeGenesisMem(out io.Writer, template genesis.Template) func(bs dtypes.ChainBlockstore, syscalls vm.SyscallBuilder) modules.Genesis {
-	return func(bs dtypes.ChainBlockstore, syscalls vm.SyscallBuilder) modules.Genesis {
+func MakeGenesisMem(out io.Writer, template genesis.Template) func(bs dtypes.ChainBlockstore, syscalls vm.SyscallBuilder, j journal.Journal) modules.Genesis {
+	return func(bs dtypes.ChainBlockstore, syscalls vm.SyscallBuilder, j journal.Journal) modules.Genesis {
 		return func() (*types.BlockHeader, error) {
 			glog.Warn("Generating new random genesis block, note that this SHOULD NOT happen unless you are setting up new network")
-			b, err := genesis2.MakeGenesisBlock(context.TODO(), bs, syscalls, template)
+			b, err := genesis2.MakeGenesisBlock(context.TODO(), j, bs, syscalls, template)
 			if err != nil {
 				return nil, xerrors.Errorf("make genesis block failed: %w", err)
 			}
@@ -50,8 +51,8 @@ func MakeGenesisMem(out io.Writer, template genesis.Template) func(bs dtypes.Cha
 	}
 }
 
-func MakeGenesis(outFile, genesisTemplate string) func(bs dtypes.ChainBlockstore, syscalls vm.SyscallBuilder) modules.Genesis {
-	return func(bs dtypes.ChainBlockstore, syscalls vm.SyscallBuilder) modules.Genesis {
+func MakeGenesis(outFile, genesisTemplate string) func(bs dtypes.ChainBlockstore, syscalls vm.SyscallBuilder, j journal.Journal) modules.Genesis {
+	return func(bs dtypes.ChainBlockstore, syscalls vm.SyscallBuilder, j journal.Journal) modules.Genesis {
 		return func() (*types.BlockHeader, error) {
 			glog.Warn("Generating new random genesis block, note that this SHOULD NOT happen unless you are setting up new network")
 			genesisTemplate, err := homedir.Expand(genesisTemplate)
@@ -73,7 +74,7 @@ func MakeGenesis(outFile, genesisTemplate string) func(bs dtypes.ChainBlockstore
 				template.Timestamp = uint64(build.Clock.Now().Unix())
 			}
 
-			b, err := genesis2.MakeGenesisBlock(context.TODO(), bs, syscalls, template)
+			b, err := genesis2.MakeGenesisBlock(context.TODO(), j, bs, syscalls, template)
 			if err != nil {
 				return nil, xerrors.Errorf("make genesis block: %w", err)
 			}

node/node_test.go

@@ -117,6 +117,16 @@ func TestPledgeSectors(t *testing.T) {
 	})
 }
 
+func TestTapeFix(t *testing.T) {
+	logging.SetLogLevel("miner", "ERROR")
+	logging.SetLogLevel("chainstore", "ERROR")
+	logging.SetLogLevel("chain", "ERROR")
+	logging.SetLogLevel("sub", "ERROR")
+	logging.SetLogLevel("storageminer", "ERROR")
+
+	test.TestTapeFix(t, builder.MockSbBuilder, 2*time.Millisecond)
+}
+
 func TestWindowedPost(t *testing.T) {
 	if os.Getenv("LOTUS_TEST_WINDOW_POST") != "1" {
 		t.Skip("this takes a few minutes, set LOTUS_TEST_WINDOW_POST=1 to run")

node/repo/fsrepo.go

@@ -44,6 +44,7 @@ const (
 	FullNode RepoType = iota
 	StorageMiner
 	Worker
+	Wallet
 )
 
 func defConfForType(t RepoType) interface{} {
@@ -54,6 +55,8 @@ func defConfForType(t RepoType) interface{} {
 		return config.DefaultStorageMiner()
 	case Worker:
 		return &struct{}{}
+	case Wallet:
+		return &struct{}{}
 	default:
 		panic(fmt.Sprintf("unknown RepoType(%d)", int(t)))
 	}
@@ -93,6 +96,12 @@ func (fsr *FsRepo) Exists() (bool, error) {
 	notexist := os.IsNotExist(err)
 	if notexist {
 		err = nil
+
+		_, err = os.Stat(filepath.Join(fsr.path, fsKeystore))
+		notexist = os.IsNotExist(err)
+		if notexist {
+			err = nil
+		}
 	}
 	return !notexist, err
 }

node/test/builder.go

@@ -7,10 +7,13 @@ import (
 	"io/ioutil"
 	"net"
 	"net/http/httptest"
+	"strings"
 	"sync"
 	"testing"
 	"time"
 
+	"golang.org/x/xerrors"
+
 	"github.com/filecoin-project/go-address"
 	"github.com/filecoin-project/go-jsonrpc"
 	"github.com/filecoin-project/go-state-types/abi"
@@ -101,8 +104,7 @@ func CreateTestStorageNode(ctx context.Context, t *testing.T, waddr address.Addr
 	var minerapi api.StorageMiner
 
 	mineBlock := make(chan miner2.MineReq)
-	// TODO: use stop
-	_, err = node.New(ctx,
+	stop, err := node.New(ctx,
 		node.StorageMiner(&minerapi),
 		node.Online(),
 		node.Repo(r),
@@ -119,6 +121,8 @@ func CreateTestStorageNode(ctx context.Context, t *testing.T, waddr address.Addr
 		t.Fatalf("failed to construct node: %v", err)
 	}
 
+	t.Cleanup(func() { _ = stop(context.Background()) })
+
 	/*// Bootstrap with full node
 	remoteAddrs, err := tnd.NetAddrsListen(ctx)
 	require.NoError(t, err)
@@ -137,11 +141,29 @@ func CreateTestStorageNode(ctx context.Context, t *testing.T, waddr address.Addr
 	return test.TestStorageNode{StorageMiner: minerapi, MineOne: mineOne}
 }
 
-func Builder(t *testing.T, nFull int, storage []test.StorageMiner, opts ...node.Option) ([]test.TestNode, []test.TestStorageNode) {
-	ctx := context.Background()
+func Builder(t *testing.T, fullOpts []test.FullNodeOpts, storage []test.StorageMiner) ([]test.TestNode, []test.TestStorageNode) {
+	return mockBuilderOpts(t, fullOpts, storage, false)
+}
+
+func MockSbBuilder(t *testing.T, fullOpts []test.FullNodeOpts, storage []test.StorageMiner) ([]test.TestNode, []test.TestStorageNode) {
+	return mockSbBuilderOpts(t, fullOpts, storage, false)
+}
+
+func RPCBuilder(t *testing.T, fullOpts []test.FullNodeOpts, storage []test.StorageMiner) ([]test.TestNode, []test.TestStorageNode) {
+	return mockBuilderOpts(t, fullOpts, storage, true)
+}
+
+func RPCMockSbBuilder(t *testing.T, fullOpts []test.FullNodeOpts, storage []test.StorageMiner) ([]test.TestNode, []test.TestStorageNode) {
+	return mockSbBuilderOpts(t, fullOpts, storage, true)
+}
+
+func mockBuilderOpts(t *testing.T, fullOpts []test.FullNodeOpts, storage []test.StorageMiner, rpc bool) ([]test.TestNode, []test.TestStorageNode) {
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+
 	mn := mocknet.New(ctx)
 
-	fulls := make([]test.TestNode, nFull)
+	fulls := make([]test.TestNode, len(fullOpts))
 	storers := make([]test.TestStorageNode, len(storage))
 
 	pk, _, err := crypto.GenerateEd25519Key(rand.Reader)
@@ -206,7 +228,7 @@ func Builder(t *testing.T, nFull int, storage []test.StorageMiner, opts ...node.
 
 	// END PRESEAL SECTION
 
-	for i := 0; i < nFull; i++ {
+	for i := 0; i < len(fullOpts); i++ {
 		var genesis node.Option
 		if i == 0 {
 			genesis = node.Override(new(modules.Genesis), testing2.MakeGenesisMem(&genbuf, *templ))
@@ -214,22 +236,26 @@ func Builder(t *testing.T, nFull int, storage []test.StorageMiner, opts ...node.
 			genesis = node.Override(new(modules.Genesis), modules.LoadGenesis(genbuf.Bytes()))
 		}
 
-		var err error
-		// TODO: Don't ignore stop
-		_, err = node.New(ctx,
-			node.FullAPI(&fulls[i].FullNode),
+		stop, err := node.New(ctx,
+			node.FullAPI(&fulls[i].FullNode, node.Lite(fullOpts[i].Lite)),
 			node.Online(),
 			node.Repo(repo.NewMemory(nil)),
 			node.MockHost(mn),
 			node.Test(),
 
 			genesis,
-			node.Options(opts...),
+
+			fullOpts[i].Opts(fulls),
 		)
 		if err != nil {
 			t.Fatal(err)
 		}
 
+		t.Cleanup(func() { _ = stop(context.Background()) })
+
+		if rpc {
+			fulls[i] = fullRpc(t, fulls[i])
+		}
 	}
 
 	for i, def := range storage {
@@ -261,6 +287,9 @@ func Builder(t *testing.T, nFull int, storage []test.StorageMiner, opts ...node.
 
 			psd := presealDirs[i]
 		*/
+		if rpc {
+			storers[i] = storerRpc(t, storers[i])
+		}
 	}
 
 	if err := mn.LinkAll(); err != nil {
@@ -286,11 +315,13 @@ func Builder(t *testing.T, nFull int, storage []test.StorageMiner, opts ...node.
 	return fulls, storers
 }
 
-func MockSbBuilder(t *testing.T, nFull int, storage []test.StorageMiner, options ...node.Option) ([]test.TestNode, []test.TestStorageNode) {
-	ctx := context.Background()
+func mockSbBuilderOpts(t *testing.T, fullOpts []test.FullNodeOpts, storage []test.StorageMiner, rpc bool) ([]test.TestNode, []test.TestStorageNode) {
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+
 	mn := mocknet.New(ctx)
 
-	fulls := make([]test.TestNode, nFull)
+	fulls := make([]test.TestNode, len(fullOpts))
 	storers := make([]test.TestStorageNode, len(storage))
 
 	var genbuf bytes.Buffer
@@ -354,7 +385,7 @@ func MockSbBuilder(t *testing.T, nFull int, storage []test.StorageMiner, options
 
 	// END PRESEAL SECTION
 
-	for i := 0; i < nFull; i++ {
+	for i := 0; i < len(fullOpts); i++ {
 		var genesis node.Option
 		if i == 0 {
 			genesis = node.Override(new(modules.Genesis), testing2.MakeGenesisMem(&genbuf, *templ))
@@ -362,10 +393,8 @@ func MockSbBuilder(t *testing.T, nFull int, storage []test.StorageMiner, options
 			genesis = node.Override(new(modules.Genesis), modules.LoadGenesis(genbuf.Bytes()))
 		}
 
-		var err error
-		// TODO: Don't ignore stop
-		_, err = node.New(ctx,
-			node.FullAPI(&fulls[i].FullNode),
+		stop, err := node.New(ctx,
+			node.FullAPI(&fulls[i].FullNode, node.Lite(fullOpts[i].Lite)),
 			node.Online(),
 			node.Repo(repo.NewMemory(nil)),
 			node.MockHost(mn),
@@ -374,11 +403,18 @@ func MockSbBuilder(t *testing.T, nFull int, storage []test.StorageMiner, options
 			node.Override(new(ffiwrapper.Verifier), mock.MockVerifier),
 
 			genesis,
-			node.Options(options...),
+
+			fullOpts[i].Opts(fulls),
 		)
 		if err != nil {
 			t.Fatalf("%+v", err)
 		}
+
+		t.Cleanup(func() { _ = stop(context.Background()) })
+
+		if rpc {
+			fulls[i] = fullRpc(t, fulls[i])
+		}
 	}
 
 	for i, def := range storage {
@@ -413,6 +449,10 @@ func MockSbBuilder(t *testing.T, nFull int, storage []test.StorageMiner, options
 			node.Override(new(ffiwrapper.Verifier), mock.MockVerifier),
 			node.Unset(new(*sectorstorage.Manager)),
 		))
+
+		if rpc {
+			storers[i] = storerRpc(t, storers[i])
+		}
 	}
 
 	if err := mn.LinkAll(); err != nil {
@@ -437,69 +477,66 @@ func MockSbBuilder(t *testing.T, nFull int, storage []test.StorageMiner, options
 	return fulls, storers
 }
 
-func RPCBuilder(t *testing.T, nFull int, storage []test.StorageMiner, opts ...node.Option) ([]test.TestNode, []test.TestStorageNode) {
-	return rpcWithBuilder(t, Builder, nFull, storage, opts...)
+func fullRpc(t *testing.T, nd test.TestNode) test.TestNode {
+	ma, listenAddr, err := CreateRPCServer(nd)
+	require.NoError(t, err)
+
+	var full test.TestNode
+	full.FullNode, _, err = client.NewFullNodeRPC(context.Background(), listenAddr, nil)
+	require.NoError(t, err)
+
+	full.ListenAddr = ma
+	return full
 }
 
-func RPCMockSbBuilder(t *testing.T, nFull int, storage []test.StorageMiner, opts ...node.Option) ([]test.TestNode, []test.TestStorageNode) {
-	return rpcWithBuilder(t, MockSbBuilder, nFull, storage, opts...)
+func storerRpc(t *testing.T, nd test.TestStorageNode) test.TestStorageNode {
+	ma, listenAddr, err := CreateRPCServer(nd)
+	require.NoError(t, err)
+
+	var storer test.TestStorageNode
+	storer.StorageMiner, _, err = client.NewStorageMinerRPC(context.Background(), listenAddr, nil)
+	require.NoError(t, err)
+
+	storer.ListenAddr = ma
+	storer.MineOne = nd.MineOne
+	return storer
 }
 
-func rpcWithBuilder(t *testing.T, b test.APIBuilder, nFull int, storage []test.StorageMiner, opts ...node.Option) ([]test.TestNode, []test.TestStorageNode) {
-	fullApis, storaApis := b(t, nFull, storage, opts...)
-	fulls := make([]test.TestNode, nFull)
-	storers := make([]test.TestStorageNode, len(storage))
+func CreateRPCServer(handler interface{}) (multiaddr.Multiaddr, string, error) {
+	rpcServer := jsonrpc.NewServer()
+	rpcServer.Register("Filecoin", handler)
+	testServ := httptest.NewServer(rpcServer) //  todo: close
 
-	for i, a := range fullApis {
-		rpcServer := jsonrpc.NewServer()
-		rpcServer.Register("Filecoin", a)
-		testServ := httptest.NewServer(rpcServer) //  todo: close
-
-		addr := testServ.Listener.Addr()
-		listenAddr := "ws://" + addr.String()
-		var err error
-		fulls[i].FullNode, _, err = client.NewFullNodeRPC(context.Background(), listenAddr, nil)
-		if err != nil {
-			t.Fatal(err)
-		}
-		ma, err := parseWSSMultiAddr(addr)
-		if err != nil {
-			t.Fatal(err)
-		}
-		fulls[i].ListenAddr = ma
+	addr := testServ.Listener.Addr()
+	listenAddr := "ws://" + addr.String()
+	ma, err := parseWSMultiAddr(addr)
+	if err != nil {
+		return nil, "", err
 	}
-
-	for i, a := range storaApis {
-		rpcServer := jsonrpc.NewServer()
-		rpcServer.Register("Filecoin", a)
-		testServ := httptest.NewServer(rpcServer) //  todo: close
-
-		addr := testServ.Listener.Addr()
-		listenAddr := "ws://" + addr.String()
-		var err error
-		storers[i].StorageMiner, _, err = client.NewStorageMinerRPC(context.Background(), listenAddr, nil)
-		if err != nil {
-			t.Fatal(err)
-		}
-		ma, err := parseWSSMultiAddr(addr)
-		if err != nil {
-			t.Fatal(err)
-		}
-		storers[i].ListenAddr = ma
-		storers[i].MineOne = a.MineOne
-	}
-
-	return fulls, storers
+	return ma, listenAddr, err
 }
 
-func parseWSSMultiAddr(addr net.Addr) (multiaddr.Multiaddr, error) {
+func parseWSMultiAddr(addr net.Addr) (multiaddr.Multiaddr, error) {
 	host, port, err := net.SplitHostPort(addr.String())
 	if err != nil {
 		return nil, err
 	}
-	ma, err := multiaddr.NewMultiaddr("/ip4/" + host + "/" + addr.Network() + "/" + port + "/wss")
+	ma, err := multiaddr.NewMultiaddr("/ip4/" + host + "/" + addr.Network() + "/" + port + "/ws")
 	if err != nil {
 		return nil, err
 	}
 	return ma, nil
 }
+
+func WSMultiAddrToString(addr multiaddr.Multiaddr) (string, error) {
+	parts := strings.Split(addr.String(), "/")
+	if len(parts) != 6 || parts[0] != "" {
+		return "", xerrors.Errorf("Malformed ws multiaddr %s", addr)
+	}
+
+	host := parts[2]
+	port := parts[4]
+	proto := parts[5]
+
+	return proto + "://" + host + ":" + port + "/rpc/v0", nil
+}