Merge branch 'master' into inmem-journal

node/builder.go

@@ -40,6 +40,10 @@ import (
 	"github.com/filecoin-project/lotus/chain/types"
 	"github.com/filecoin-project/lotus/chain/vm"
 	"github.com/filecoin-project/lotus/chain/wallet"
+	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"
+	sealing "github.com/filecoin-project/lotus/extern/storage-sealing"
 	"github.com/filecoin-project/lotus/journal"
 	"github.com/filecoin-project/lotus/lib/blockstore"
 	"github.com/filecoin-project/lotus/lib/peermgr"
@@ -62,12 +66,9 @@ import (
 	"github.com/filecoin-project/lotus/paychmgr/settler"
 	"github.com/filecoin-project/lotus/storage"
 	"github.com/filecoin-project/lotus/storage/sectorblocks"
-	sectorstorage "github.com/filecoin-project/sector-storage"
-	"github.com/filecoin-project/sector-storage/ffiwrapper"
-	"github.com/filecoin-project/sector-storage/stores"
-	sealing "github.com/filecoin-project/storage-fsm"
 )
 
+//nolint:deadcode,varcheck
 var log = logging.Logger("builder")
 
 // special is a type used to give keys to modules which
@@ -109,7 +110,6 @@ const (
 	HandleIncomingBlocksKey
 	HandleIncomingMessagesKey
 
-	RegisterClientValidatorKey
 	HandlePaymentChannelManagerKey
 
 	// miner
@@ -117,7 +117,6 @@ const (
 	HandleDealsKey
 	HandleRetrievalKey
 	RunSectorServiceKey
-	RegisterProviderValidatorKey
 
 	// daemon
 	ExtractApiKey
@@ -183,7 +182,7 @@ func libp2p() Option {
 		Override(AddrsFactoryKey, lp2p.AddrsFactory(nil, nil)),
 		Override(SmuxTransportKey, lp2p.SmuxTransport(true)),
 		Override(RelayKey, lp2p.NoRelay()),
-		Override(SecurityKey, lp2p.Security(true, true)),
+		Override(SecurityKey, lp2p.Security(true, false)),
 
 		Override(BaseRoutingKey, lp2p.BaseRouting),
 		Override(new(routing.Routing), lp2p.Routing),
@@ -262,6 +261,7 @@ func Online() Option {
 			Override(new(*peermgr.PeerMgr), peermgr.NewPeerMgr),
 
 			Override(new(dtypes.Graphsync), modules.Graphsync),
+			Override(new(*dtypes.MpoolLocker), new(dtypes.MpoolLocker)),
 
 			Override(RunHelloKey, modules.RunHello),
 			Override(RunBlockSyncKey, modules.RunBlockSync),
@@ -272,14 +272,11 @@ func Online() Option {
 			Override(new(retrievalmarket.PeerResolver), modules.RetrievalResolver),
 
 			Override(new(retrievalmarket.RetrievalClient), modules.RetrievalClient),
-			Override(new(dtypes.ClientDealStore), modules.NewClientDealStore),
 			Override(new(dtypes.ClientDatastore), modules.NewClientDatastore),
 			Override(new(dtypes.ClientDataTransfer), modules.NewClientGraphsyncDataTransfer),
-			Override(new(dtypes.ClientRequestValidator), modules.NewClientRequestValidator),
 			Override(new(modules.ClientDealFunds), modules.NewClientDealFunds),
 			Override(new(storagemarket.StorageClient), modules.StorageClient),
 			Override(new(storagemarket.StorageClientNode), storageadapter.NewClientNodeAdapter),
-			Override(RegisterClientValidatorKey, modules.RegisterClientValidator),
 			Override(new(beacon.RandomBeacon), modules.RandomBeacon),
 
 			Override(new(*paychmgr.Store), paychmgr.NewStore),
@@ -308,7 +305,7 @@ func Online() Option {
 			Override(new(storage2.Prover), From(new(sectorstorage.SectorManager))),
 
 			Override(new(*sectorblocks.SectorBlocks), sectorblocks.NewSectorBlocks),
-			Override(new(*storage.Miner), modules.StorageMiner),
+			Override(new(*storage.Miner), modules.StorageMiner(config.DefaultStorageMiner().Fees)),
 			Override(new(dtypes.NetworkName), modules.StorageNetworkName),
 
 			Override(new(dtypes.StagingMultiDstore), modules.StagingMultiDatastore),
@@ -316,16 +313,13 @@ func Online() Option {
 			Override(new(dtypes.StagingDAG), modules.StagingDAG),
 			Override(new(dtypes.StagingGraphsync), modules.StagingGraphsync),
 			Override(new(retrievalmarket.RetrievalProvider), modules.RetrievalProvider),
-			Override(new(dtypes.ProviderDealStore), modules.NewProviderDealStore),
 			Override(new(dtypes.ProviderDataTransfer), modules.NewProviderDAGServiceDataTransfer),
-			Override(new(dtypes.ProviderRequestValidator), modules.NewProviderRequestValidator),
 			Override(new(dtypes.ProviderPieceStore), modules.NewProviderPieceStore),
 			Override(new(*storedask.StoredAsk), modules.NewStorageAsk),
 			Override(new(dtypes.DealFilter), modules.BasicDealFilter(nil)),
 			Override(new(modules.ProviderDealFunds), modules.NewProviderDealFunds),
 			Override(new(storagemarket.StorageProvider), modules.StorageProvider),
 			Override(new(storagemarket.StorageProviderNode), storageadapter.NewProviderNodeAdapter),
-			Override(RegisterProviderValidatorKey, modules.RegisterProviderValidator),
 			Override(HandleRetrievalKey, modules.HandleRetrieval),
 			Override(GetParamsKey, modules.GetParams),
 			Override(HandleDealsKey, modules.HandleDeals),
@@ -342,8 +336,8 @@ func Online() Option {
 			Override(new(dtypes.SetConsiderOfflineStorageDealsConfigFunc), modules.NewSetConsideringOfflineStorageDealsFunc),
 			Override(new(dtypes.ConsiderOfflineRetrievalDealsConfigFunc), modules.NewConsiderOfflineRetrievalDealsConfigFunc),
 			Override(new(dtypes.SetConsiderOfflineRetrievalDealsConfigFunc), modules.NewSetConsiderOfflineRetrievalDealsConfigFunc),
-			Override(new(dtypes.SetSealingDelayFunc), modules.NewSetSealDelayFunc),
-			Override(new(dtypes.GetSealingDelayFunc), modules.NewGetSealDelayFunc),
+			Override(new(dtypes.SetSealingConfigFunc), modules.NewSetSealConfigFunc),
+			Override(new(dtypes.GetSealingConfigFunc), modules.NewGetSealConfigFunc),
 			Override(new(dtypes.SetExpectedSealDurationFunc), modules.NewSetExpectedSealDurationFunc),
 			Override(new(dtypes.GetExpectedSealDurationFunc), modules.NewGetExpectedSealDurationFunc),
 		),
@@ -445,6 +439,7 @@ func ConfigStorageMiner(c interface{}) Option {
 		),
 
 		Override(new(sectorstorage.SealerConfig), cfg.Storage),
+		Override(new(*storage.Miner), modules.StorageMiner(cfg.Fees)),
 	)
 }
 

node/config/def.go

@@ -6,7 +6,8 @@ import (
 
 	"github.com/ipfs/go-cid"
 
-	sectorstorage "github.com/filecoin-project/sector-storage"
+	"github.com/filecoin-project/lotus/chain/types"
+	sectorstorage "github.com/filecoin-project/lotus/extern/sector-storage"
 )
 
 // Common is common config between full node and miner
@@ -29,9 +30,10 @@ type FullNode struct {
 type StorageMiner struct {
 	Common
 
-	Dealmaking   DealmakingConfig
-	Storage      sectorstorage.SealerConfig
-	SealingDelay Duration
+	Dealmaking DealmakingConfig
+	Sealing    SealingConfig
+	Storage    sectorstorage.SealerConfig
+	Fees       MinerFeeConfig
 }
 
 type DealmakingConfig struct {
@@ -45,6 +47,25 @@ type DealmakingConfig struct {
 	Filter string
 }
 
+type SealingConfig struct {
+	// 0 = no limit
+	MaxWaitDealsSectors uint64
+
+	// includes failed, 0 = no limit
+	MaxSealingSectors uint64
+
+	// includes failed, 0 = no limit
+	MaxSealingSectorsForDeals uint64
+
+	WaitDealsDelay Duration
+}
+
+type MinerFeeConfig struct {
+	MaxPreCommitGasFee  types.FIL
+	MaxCommitGasFee     types.FIL
+	MaxWindowPoStGasFee types.FIL
+}
+
 // API contains configs for API endpoint
 type API struct {
 	ListenAddress       string
@@ -105,7 +126,7 @@ func defCommon() Common {
 		Pubsub: Pubsub{
 			Bootstrapper: false,
 			DirectPeers:  nil,
-			RemoteTracer: "/ip4/147.75.67.199/tcp/4001/p2p/QmTd6UvR47vUidRNZ1ZKXHrAFhqTJAD27rKL9XYghEKgKX",
+			RemoteTracer: "/dns4/pubsub-tracer.filecoin.io/tcp/4001/p2p/QmTd6UvR47vUidRNZ1ZKXHrAFhqTJAD27rKL9XYghEKgKX",
 		},
 	}
 
@@ -122,7 +143,15 @@ func DefaultStorageMiner() *StorageMiner {
 	cfg := &StorageMiner{
 		Common: defCommon(),
 
+		Sealing: SealingConfig{
+			MaxWaitDealsSectors:       2, // 64G with 32G sectors
+			MaxSealingSectors:         0,
+			MaxSealingSectorsForDeals: 0,
+			WaitDealsDelay:            Duration(time.Hour),
+		},
+
 		Storage: sectorstorage.SealerConfig{
+			AllowAddPiece:   true,
 			AllowPreCommit1: true,
 			AllowPreCommit2: true,
 			AllowCommit:     true,
@@ -143,7 +172,11 @@ func DefaultStorageMiner() *StorageMiner {
 			ExpectedSealDuration: Duration(time.Hour * 12),
 		},
 
-		SealingDelay: Duration(time.Hour),
+		Fees: MinerFeeConfig{
+			MaxPreCommitGasFee:  types.FIL(types.BigDiv(types.FromFil(1), types.NewInt(20))), // 0.05
+			MaxCommitGasFee:     types.FIL(types.BigDiv(types.FromFil(1), types.NewInt(20))),
+			MaxWindowPoStGasFee: types.FIL(types.FromFil(50)),
+		},
 	}
 	cfg.Common.API.ListenAddress = "/ip4/127.0.0.1/tcp/2345/http"
 	cfg.Common.API.RemoteListenAddress = "127.0.0.1:2345"

node/config/def_test.go

@@ -0,0 +1,54 @@
+package config
+
+import (
+	"bytes"
+	"fmt"
+	"reflect"
+	"strings"
+	"testing"
+
+	"github.com/BurntSushi/toml"
+	"github.com/stretchr/testify/require"
+)
+
+func TestDefaultFullNodeRoundtrip(t *testing.T) {
+	c := DefaultFullNode()
+
+	var s string
+	{
+		buf := new(bytes.Buffer)
+		_, _ = buf.WriteString("# Default config:\n")
+		e := toml.NewEncoder(buf)
+		require.NoError(t, e.Encode(c))
+
+		s = buf.String()
+	}
+
+	c2, err := FromReader(strings.NewReader(s), DefaultFullNode())
+	require.NoError(t, err)
+
+	fmt.Println(s)
+
+	require.True(t, reflect.DeepEqual(c, c2))
+}
+
+func TestDefaultMinerRoundtrip(t *testing.T) {
+	c := DefaultStorageMiner()
+
+	var s string
+	{
+		buf := new(bytes.Buffer)
+		_, _ = buf.WriteString("# Default config:\n")
+		e := toml.NewEncoder(buf)
+		require.NoError(t, e.Encode(c))
+
+		s = buf.String()
+	}
+
+	c2, err := FromReader(strings.NewReader(s), DefaultStorageMiner())
+	require.NoError(t, err)
+
+	fmt.Println(s)
+
+	require.True(t, reflect.DeepEqual(c, c2))
+}

node/config/load.go

@@ -53,5 +53,4 @@ func ConfigComment(t interface{}) ([]byte, error) {
 	b = bytes.ReplaceAll(b, []byte("\n"), []byte("\n#"))
 	b = bytes.ReplaceAll(b, []byte("#["), []byte("["))
 	return b, nil
-
 }

node/config/storage.go

@@ -8,7 +8,7 @@ import (
 
 	"golang.org/x/xerrors"
 
-	"github.com/filecoin-project/sector-storage/stores"
+	"github.com/filecoin-project/lotus/extern/sector-storage/stores"
 )
 
 func StorageFromFile(path string, def *stores.StorageConfig) (*stores.StorageConfig, error) {

node/hello/cbor_gen.go

@@ -6,7 +6,7 @@ import (
 	"fmt"
 	"io"
 
-	"github.com/filecoin-project/specs-actors/actors/abi"
+	abi "github.com/filecoin-project/specs-actors/actors/abi"
 	cid "github.com/ipfs/go-cid"
 	cbg "github.com/whyrusleeping/cbor-gen"
 	xerrors "golang.org/x/xerrors"

node/impl/client/client.go

@@ -6,6 +6,8 @@ import (
 	"io"
 	"os"
 
+	datatransfer "github.com/filecoin-project/go-data-transfer"
+	"github.com/filecoin-project/specs-actors/actors/abi/big"
 	"golang.org/x/xerrors"
 
 	"github.com/ipfs/go-blockservice"
@@ -23,6 +25,7 @@ import (
 	basicnode "github.com/ipld/go-ipld-prime/node/basic"
 	"github.com/ipld/go-ipld-prime/traversal/selector"
 	"github.com/ipld/go-ipld-prime/traversal/selector/builder"
+	"github.com/libp2p/go-libp2p-core/host"
 	"github.com/libp2p/go-libp2p-core/peer"
 	mh "github.com/multiformats/go-multihash"
 	"go.uber.org/fx"
@@ -35,11 +38,12 @@ import (
 	"github.com/filecoin-project/go-fil-markets/storagemarket"
 	"github.com/filecoin-project/go-multistore"
 	"github.com/filecoin-project/go-padreader"
-	"github.com/filecoin-project/sector-storage/ffiwrapper"
 	"github.com/filecoin-project/specs-actors/actors/abi"
-	"github.com/filecoin-project/specs-actors/actors/abi/big"
 	"github.com/filecoin-project/specs-actors/actors/builtin/miner"
 
+	"github.com/filecoin-project/lotus/extern/sector-storage/ffiwrapper"
+	marketevents "github.com/filecoin-project/lotus/markets/loggers"
+
 	"github.com/filecoin-project/lotus/api"
 	"github.com/filecoin-project/lotus/build"
 	"github.com/filecoin-project/lotus/chain/store"
@@ -72,6 +76,8 @@ type API struct {
 
 	CombinedBstore    dtypes.ClientBlockstore // TODO: try to remove
 	RetrievalStoreMgr dtypes.ClientRetrievalStoreManager
+	DataTransfer      dtypes.ClientDataTransfer
+	Host              host.Host
 }
 
 func calcDealExpiration(minDuration uint64, md *miner.DeadlineInfo, startEpoch abi.ChainEpoch) abi.ChainEpoch {
@@ -103,7 +109,7 @@ func (a *API) ClientStartDeal(ctx context.Context, params *api.StartDealParams)
 				continue
 			}
 			if c.Equals(params.Data.Root) {
-				storeID = &importID
+				storeID = &importID //nolint
 				break
 			}
 		}
@@ -135,7 +141,7 @@ 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)
+	providerInfo := utils.NewStorageProviderInfo(params.Miner, mi.Worker, mi.SectorSize, *mi.PeerId, mi.Multiaddrs)
 
 	dealStart := params.DealStartEpoch
 	if dealStart <= 0 { // unset, or explicitly 'epoch undefined'
@@ -155,7 +161,7 @@ func (a *API) ClientStartDeal(ctx context.Context, params *api.StartDealParams)
 		StartEpoch:    dealStart,
 		EndEpoch:      calcDealExpiration(params.MinBlocksDuration, md, dealStart),
 		Price:         params.EpochPrice,
-		Collateral:    big.Zero(),
+		Collateral:    params.ProviderCollateral,
 		Rt:            rt,
 		FastRetrieval: params.FastRetrieval,
 		VerifiedDeal:  params.VerifiedDeal,
@@ -190,6 +196,7 @@ func (a *API) ClientListDeals(ctx context.Context) ([]api.DealInfo, error) {
 			PricePerEpoch: v.Proposal.StoragePricePerEpoch,
 			Duration:      uint64(v.Proposal.Duration()),
 			DealID:        v.DealID,
+			CreationTime:  v.CreationTime.Time(),
 		}
 	}
 
@@ -212,6 +219,7 @@ func (a *API) ClientGetDealInfo(ctx context.Context, d cid.Cid) (*api.DealInfo,
 		PricePerEpoch: v.Proposal.StoragePricePerEpoch,
 		Duration:      uint64(v.Proposal.Duration()),
 		DealID:        v.DealID,
+		CreationTime:  v.CreationTime.Time(),
 	}, nil
 }
 
@@ -253,7 +261,7 @@ func (a *API) ClientMinerQueryOffer(ctx context.Context, miner address.Address,
 	}
 	rp := rm.RetrievalPeer{
 		Address: miner,
-		ID:      mi.PeerId,
+		ID:      *mi.PeerId,
 	}
 	return a.makeRetrievalQuery(ctx, rp, root, piece, rm.QueryParams{}), nil
 }
@@ -399,20 +407,56 @@ func (a *API) ClientListImports(ctx context.Context) ([]api.Import, error) {
 }
 
 func (a *API) ClientRetrieve(ctx context.Context, order api.RetrievalOrder, ref *api.FileRef) error {
+	events := make(chan marketevents.RetrievalEvent)
+	go a.clientRetrieve(ctx, order, ref, events)
+
+	for {
+		select {
+		case evt, ok := <-events:
+			if !ok { // done successfully
+				return nil
+			}
+
+			if evt.Err != "" {
+				return xerrors.Errorf("retrieval failed: %s", evt.Err)
+			}
+		case <-ctx.Done():
+			return xerrors.Errorf("retrieval timed out")
+		}
+	}
+}
+
+func (a *API) ClientRetrieveWithEvents(ctx context.Context, order api.RetrievalOrder, ref *api.FileRef) (<-chan marketevents.RetrievalEvent, error) {
+	events := make(chan marketevents.RetrievalEvent)
+	go a.clientRetrieve(ctx, order, ref, events)
+	return events, nil
+}
+
+func (a *API) clientRetrieve(ctx context.Context, order api.RetrievalOrder, ref *api.FileRef, events chan marketevents.RetrievalEvent) {
+	defer close(events)
+
+	finish := func(e error) {
+		if e != nil {
+			events <- marketevents.RetrievalEvent{Err: e.Error(), FundsSpent: big.Zero()}
+		}
+	}
+
 	if order.MinerPeer.ID == "" {
 		mi, err := a.StateMinerInfo(ctx, order.Miner, types.EmptyTSK)
 		if err != nil {
-			return err
+			finish(err)
+			return
 		}
 
 		order.MinerPeer = retrievalmarket.RetrievalPeer{
-			ID:      mi.PeerId,
+			ID:      *mi.PeerId,
 			Address: order.Miner,
 		}
 	}
 
 	if order.Size == 0 {
-		return xerrors.Errorf("cannot make retrieval deal for zero bytes")
+		finish(xerrors.Errorf("cannot make retrieval deal for zero bytes"))
+		return
 	}
 
 	/*id, st, err := a.imgr().NewStore()
@@ -425,8 +469,22 @@ func (a *API) ClientRetrieve(ctx context.Context, order api.RetrievalOrder, ref
 
 	retrievalResult := make(chan error, 1)
 
+	var dealId retrievalmarket.DealID
+
 	unsubscribe := a.Retrieval.SubscribeToEvents(func(event rm.ClientEvent, state rm.ClientDealState) {
-		if state.PayloadCID.Equals(order.Root) {
+		if state.PayloadCID.Equals(order.Root) && state.ID == dealId {
+
+			select {
+			case <-ctx.Done():
+				return
+			case events <- marketevents.RetrievalEvent{
+				Event:         event,
+				Status:        state.Status,
+				BytesReceived: state.TotalReceived,
+				FundsSpent:    state.FundsSpent,
+			}:
+			}
+
 			switch state.Status {
 			case rm.DealStatusCompleted:
 				retrievalResult <- nil
@@ -444,19 +502,21 @@ func (a *API) ClientRetrieve(ctx context.Context, order api.RetrievalOrder, ref
 
 	params, err := rm.NewParamsV1(ppb, order.PaymentInterval, order.PaymentIntervalIncrease, shared.AllSelector(), order.Piece, order.UnsealPrice)
 	if err != nil {
-		return xerrors.Errorf("Error in retrieval params: %s", err)
+		finish(xerrors.Errorf("Error in retrieval params: %s", err))
+		return
 	}
 
 	store, err := a.RetrievalStoreMgr.NewStore()
 	if err != nil {
-		return xerrors.Errorf("Error setting up new store: %w", err)
+		finish(xerrors.Errorf("Error setting up new store: %w", err))
+		return
 	}
 
 	defer func() {
 		_ = a.RetrievalStoreMgr.ReleaseStore(store)
 	}()
 
-	_, err = a.Retrieval.Retrieve(
+	dealId, err = a.Retrieval.Retrieve(
 		ctx,
 		order.Root,
 		params,
@@ -467,22 +527,27 @@ func (a *API) ClientRetrieve(ctx context.Context, order api.RetrievalOrder, ref
 		store.StoreID())
 
 	if err != nil {
-		return xerrors.Errorf("Retrieve failed: %w", err)
+		finish(xerrors.Errorf("Retrieve failed: %w", err))
+		return
 	}
+
 	select {
 	case <-ctx.Done():
-		return xerrors.New("Retrieval Timed Out")
+		unsubscribe()
+		finish(xerrors.New("Retrieval Timed Out"))
+		return
 	case err := <-retrievalResult:
+		unsubscribe()
 		if err != nil {
-			return xerrors.Errorf("Retrieve: %w", err)
+			finish(xerrors.Errorf("Retrieve: %w", err))
+			return
 		}
 	}
 
-	unsubscribe()
-
 	// If ref is nil, it only fetches the data into the configured blockstore.
 	if ref == nil {
-		return nil
+		finish(nil)
+		return
 	}
 
 	rdag := store.DAGService()
@@ -490,24 +555,30 @@ func (a *API) ClientRetrieve(ctx context.Context, order api.RetrievalOrder, ref
 	if ref.IsCAR {
 		f, err := os.OpenFile(ref.Path, os.O_CREATE|os.O_WRONLY, 0644)
 		if err != nil {
-			return err
+			finish(err)
+			return
 		}
 		err = car.WriteCar(ctx, rdag, []cid.Cid{order.Root}, f)
 		if err != nil {
-			return err
+			finish(err)
+			return
 		}
-		return f.Close()
+		finish(f.Close())
+		return
 	}
 
 	nd, err := rdag.Get(ctx, order.Root)
 	if err != nil {
-		return xerrors.Errorf("ClientRetrieve: %w", err)
+		finish(xerrors.Errorf("ClientRetrieve: %w", err))
+		return
 	}
 	file, err := unixfile.NewUnixfsFile(ctx, rdag, nd)
 	if err != nil {
-		return xerrors.Errorf("ClientRetrieve: %w", err)
+		finish(xerrors.Errorf("ClientRetrieve: %w", err))
+		return
 	}
-	return files.WriteTo(file, ref.Path)
+	finish(files.WriteTo(file, ref.Path))
+	return
 }
 
 func (a *API) ClientQueryAsk(ctx context.Context, p peer.ID, miner address.Address) (*storagemarket.SignedStorageAsk, error) {
@@ -524,13 +595,19 @@ func (a *API) ClientQueryAsk(ctx context.Context, p peer.ID, miner address.Addre
 	return signedAsk, nil
 }
 
-func (a *API) ClientCalcCommP(ctx context.Context, inpath string, miner address.Address) (*api.CommPRet, error) {
-	mi, err := a.StateMinerInfo(ctx, miner, types.EmptyTSK)
-	if err != nil {
-		return nil, xerrors.Errorf("failed checking miners sector size: %w", err)
-	}
+func (a *API) ClientCalcCommP(ctx context.Context, inpath string) (*api.CommPRet, error) {
 
-	rt, err := ffiwrapper.SealProofTypeFromSectorSize(mi.SectorSize)
+	// Hard-code the sector size to 32GiB, because:
+	// - pieceio.GeneratePieceCommitment requires a RegisteredSealProof
+	// - commP itself is sector-size independent, with rather low probability of that changing
+	//   ( note how the final rust call is identical for every RegSP type )
+	//   https://github.com/filecoin-project/rust-filecoin-proofs-api/blob/v5.0.0/src/seal.rs#L1040-L1050
+	//
+	// IF/WHEN this changes in the future we will have to be able to calculate
+	// "old style" commP, and thus will need to introduce a version switch or similar
+	arbitrarySectorSize := abi.SectorSize(32 << 30)
+
+	rt, err := ffiwrapper.SealProofTypeFromSectorSize(arbitrarySectorSize)
 	if err != nil {
 		return nil, xerrors.Errorf("bad sector size: %w", err)
 	}
@@ -539,7 +616,7 @@ func (a *API) ClientCalcCommP(ctx context.Context, inpath string, miner address.
 	if err != nil {
 		return nil, err
 	}
-	defer rdr.Close()
+	defer rdr.Close() //nolint:errcheck
 
 	stat, err := rdr.Stat()
 	if err != nil {
@@ -625,7 +702,7 @@ func (a *API) clientImport(ctx context.Context, ref api.FileRef, store *multisto
 	if err != nil {
 		return cid.Undef, err
 	}
-	defer f.Close()
+	defer f.Close() //nolint:errcheck
 
 	stat, err := f.Stat()
 	if err != nil {
@@ -690,3 +767,36 @@ func (a *API) clientImport(ctx context.Context, ref api.FileRef, store *multisto
 
 	return nd.Cid(), nil
 }
+
+func (a *API) ClientListDataTransfers(ctx context.Context) ([]api.DataTransferChannel, error) {
+	inProgressChannels, err := a.DataTransfer.InProgressChannels(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	apiChannels := make([]api.DataTransferChannel, 0, len(inProgressChannels))
+	for _, channelState := range inProgressChannels {
+		apiChannels = append(apiChannels, api.NewDataTransferChannel(a.Host.ID(), channelState))
+	}
+
+	return apiChannels, nil
+}
+
+func (a *API) ClientDataTransferUpdates(ctx context.Context) (<-chan api.DataTransferChannel, error) {
+	channels := make(chan api.DataTransferChannel)
+
+	unsub := a.DataTransfer.SubscribeToEvents(func(evt datatransfer.Event, channelState datatransfer.ChannelState) {
+		channel := api.NewDataTransferChannel(a.Host.ID(), channelState)
+		select {
+		case <-ctx.Done():
+		case channels <- channel:
+		}
+	})
+
+	go func() {
+		defer unsub()
+		<-ctx.Done()
+	}()
+
+	return channels, nil
+}

node/impl/common/common.go

@@ -12,6 +12,7 @@ import (
 	"github.com/libp2p/go-libp2p-core/network"
 	"github.com/libp2p/go-libp2p-core/peer"
 	swarm "github.com/libp2p/go-libp2p-swarm"
+	basichost "github.com/libp2p/go-libp2p/p2p/host/basic"
 	ma "github.com/multiformats/go-multiaddr"
 	"go.uber.org/fx"
 	"golang.org/x/xerrors"
@@ -28,6 +29,7 @@ type CommonAPI struct {
 	fx.In
 
 	APISecret    *dtypes.APIAlg
+	RawHost      lp2p.RawHost
 	Host         host.Host
 	Router       lp2p.BaseIpfsRouting
 	Sk           *dtypes.ScoreKeeper
@@ -113,6 +115,24 @@ func (a *CommonAPI) NetFindPeer(ctx context.Context, p peer.ID) (peer.AddrInfo,
 	return a.Router.FindPeer(ctx, p)
 }
 
+func (a *CommonAPI) NetAutoNatStatus(ctx context.Context) (i api.NatInfo, err error) {
+	autonat := a.RawHost.(*basichost.BasicHost).AutoNat
+
+	var maddr string
+	if autonat.Status() == network.ReachabilityPublic {
+		pa, err := autonat.PublicAddr()
+		if err != nil {
+			return api.NatInfo{}, err
+		}
+		maddr = pa.String()
+	}
+
+	return api.NatInfo{
+		Reachability: autonat.Status(),
+		PublicAddr:   maddr,
+	}, nil
+}
+
 func (a *CommonAPI) ID(context.Context) (peer.ID, error) {
 	return a.Host.ID(), nil
 }

node/impl/full/chain.go

@@ -54,13 +54,22 @@ func (a *ChainAPI) ChainHead(context.Context) (*types.TipSet, error) {
 	return a.Chain.GetHeaviestTipSet(), nil
 }
 
-func (a *ChainAPI) ChainGetRandomness(ctx context.Context, tsk types.TipSetKey, personalization crypto.DomainSeparationTag, randEpoch abi.ChainEpoch, entropy []byte) (abi.Randomness, error) {
+func (a *ChainAPI) ChainGetRandomnessFromTickets(ctx context.Context, tsk types.TipSetKey, personalization crypto.DomainSeparationTag, randEpoch abi.ChainEpoch, entropy []byte) (abi.Randomness, error) {
 	pts, err := a.Chain.LoadTipSet(tsk)
 	if err != nil {
 		return nil, xerrors.Errorf("loading tipset key: %w", err)
 	}
 
-	return a.Chain.GetRandomness(ctx, pts.Cids(), personalization, randEpoch, entropy)
+	return a.Chain.GetChainRandomness(ctx, pts.Cids(), personalization, randEpoch, entropy)
+}
+
+func (a *ChainAPI) ChainGetRandomnessFromBeacon(ctx context.Context, tsk types.TipSetKey, personalization crypto.DomainSeparationTag, randEpoch abi.ChainEpoch, entropy []byte) (abi.Randomness, error) {
+	pts, err := a.Chain.LoadTipSet(tsk)
+	if err != nil {
+		return nil, xerrors.Errorf("loading tipset key: %w", err)
+	}
+
+	return a.Chain.GetBeaconRandomness(ctx, pts.Cids(), personalization, randEpoch, entropy)
 }
 
 func (a *ChainAPI) ChainGetBlock(ctx context.Context, msg cid.Cid) (*types.BlockHeader, error) {

node/impl/full/gas.go

@@ -3,8 +3,10 @@ package full
 import (
 	"context"
 	"math"
+	"math/rand"
 	"sort"
 
+	"github.com/filecoin-project/lotus/api"
 	"github.com/filecoin-project/lotus/build"
 	"github.com/filecoin-project/lotus/chain/messagepool"
 	"github.com/filecoin-project/lotus/chain/stmgr"
@@ -13,6 +15,7 @@ import (
 
 	"github.com/filecoin-project/go-address"
 	"github.com/filecoin-project/specs-actors/actors/abi/big"
+	"github.com/filecoin-project/specs-actors/actors/builtin"
 	"github.com/filecoin-project/specs-actors/actors/runtime/exitcode"
 
 	"go.uber.org/fx"
@@ -26,7 +29,7 @@ type GasAPI struct {
 	Mpool *messagepool.MessagePool
 }
 
-const MinGasPremium = 10e3
+const MinGasPremium = 100e3
 const MaxSpendOnFeeDenom = 100
 
 func (a *GasAPI) GasEstimateFeeCap(ctx context.Context, msg *types.Message, maxqueueblks int64,
@@ -76,7 +79,7 @@ func (a *GasAPI) GasEstimateGasPremium(ctx context.Context, nblocksincl uint64,
 
 	ts := a.Chain.GetHeaviestTipSet()
 	for i := uint64(0); i < nblocksincl*2; i++ {
-		if len(ts.Parents().Cids()) == 0 {
+		if ts.Height() == 0 {
 			break // genesis
 		}
 
@@ -111,6 +114,7 @@ func (a *GasAPI) GasEstimateGasPremium(ctx context.Context, nblocksincl uint64,
 	at := build.BlockGasTarget * int64(blocks) / 2
 	prev := big.Zero()
 
+	premium := big.Zero()
 	for _, price := range prices {
 		at -= price.limit
 		if at > 0 {
@@ -122,17 +126,27 @@ func (a *GasAPI) GasEstimateGasPremium(ctx context.Context, nblocksincl uint64,
 			return types.BigAdd(price.price, big.NewInt(1)), nil
 		}
 
-		return types.BigAdd(big.Div(types.BigAdd(price.price, prev), types.NewInt(2)), big.NewInt(1)), nil
+		premium = types.BigAdd(big.Div(types.BigAdd(price.price, prev), types.NewInt(2)), big.NewInt(1))
 	}
 
-	switch nblocksincl {
-	case 1:
-		return types.NewInt(2 * MinGasPremium), nil
-	case 2:
-		return types.NewInt(1.5 * MinGasPremium), nil
-	default:
-		return types.NewInt(MinGasPremium), nil
+	if types.BigCmp(premium, big.Zero()) == 0 {
+		switch nblocksincl {
+		case 1:
+			premium = types.NewInt(2 * MinGasPremium)
+		case 2:
+			premium = types.NewInt(1.5 * MinGasPremium)
+		default:
+			premium = types.NewInt(MinGasPremium)
+		}
 	}
+
+	// add some noise to normalize behaviour of message selection
+	const precision = 32
+	// mean 1, stddev 0.005 => 95% within +-1%
+	noise := 1 + rand.NormFloat64()*0.005
+	premium = types.BigMul(premium, types.NewInt(uint64(noise*(1<<precision))))
+	premium = types.BigDiv(premium, types.NewInt(1<<precision))
+	return premium, nil
 }
 
 func (a *GasAPI) GasEstimateGasLimit(ctx context.Context, msgIn *types.Message, _ types.TipSetKey) (int64, error) {
@@ -162,5 +176,53 @@ func (a *GasAPI) GasEstimateGasLimit(ctx context.Context, msgIn *types.Message,
 		return -1, xerrors.Errorf("message execution failed: exit %s, reason: %s", res.MsgRct.ExitCode, res.Error)
 	}
 
-	return res.MsgRct.GasUsed, nil
+	// Special case for PaymentChannel collect, which is deleting actor
+	var act types.Actor
+	err = a.Stmgr.WithParentState(ts, a.Stmgr.WithActor(msg.To, stmgr.GetActor(&act)))
+	if err != nil {
+		_ = err
+		// somewhat ignore it as it can happen and we just want to detect
+		// an existing PaymentChannel actor
+		return res.MsgRct.GasUsed, nil
+	}
+
+	if !act.Code.Equals(builtin.PaymentChannelActorCodeID) {
+		return res.MsgRct.GasUsed, nil
+	}
+	if msgIn.Method != builtin.MethodsPaych.Collect {
+		return res.MsgRct.GasUsed, nil
+	}
+
+	// return GasUsed without the refund for DestoryActor
+	return res.MsgRct.GasUsed + 76e3, nil
+}
+
+func (a *GasAPI) 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{})
+		if err != nil {
+			return nil, xerrors.Errorf("estimating gas used: %w", err)
+		}
+		msg.GasLimit = int64(float64(gasLimit) * a.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{})
+		if err != nil {
+			return nil, xerrors.Errorf("estimating gas price: %w", err)
+		}
+		msg.GasPremium = gasPremium
+	}
+
+	if msg.GasFeeCap == types.EmptyInt || types.BigCmp(msg.GasFeeCap, types.NewInt(0)) == 0 {
+		feeCap, err := a.GasEstimateFeeCap(ctx, msg, 10, types.EmptyTSK)
+		if err != nil {
+			return nil, xerrors.Errorf("estimating fee cap: %w", err)
+		}
+		msg.GasFeeCap = big.Add(feeCap, msg.GasPremium)
+	}
+
+	capGasFee(msg, spec.Get().MaxFee)
+
+	return msg, nil
 }

node/impl/full/mpool.go

@@ -2,17 +2,20 @@ package full
 
 import (
 	"context"
-	"sync"
 
+	"github.com/filecoin-project/go-address"
 	"github.com/ipfs/go-cid"
 	"go.uber.org/fx"
 	"golang.org/x/xerrors"
 
-	"github.com/filecoin-project/go-address"
+	"github.com/filecoin-project/specs-actors/actors/abi"
+	"github.com/filecoin-project/specs-actors/actors/abi/big"
+
 	"github.com/filecoin-project/lotus/api"
 	"github.com/filecoin-project/lotus/chain/messagepool"
 	"github.com/filecoin-project/lotus/chain/store"
 	"github.com/filecoin-project/lotus/chain/types"
+	"github.com/filecoin-project/lotus/node/modules/dtypes"
 )
 
 type MpoolAPI struct {
@@ -25,10 +28,7 @@ type MpoolAPI struct {
 
 	Mpool *messagepool.MessagePool
 
-	PushLocks struct {
-		m map[address.Address]chan struct{}
-		sync.Mutex
-	} `name:"verymuchunique" optional:"true"`
+	PushLocks *dtypes.MpoolLocker
 }
 
 func (a *MpoolAPI) MpoolGetConfig(context.Context) (*types.MpoolConfig, error) {
@@ -36,17 +36,16 @@ func (a *MpoolAPI) MpoolGetConfig(context.Context) (*types.MpoolConfig, error) {
 }
 
 func (a *MpoolAPI) MpoolSetConfig(ctx context.Context, cfg *types.MpoolConfig) error {
-	a.Mpool.SetConfig(cfg)
-	return nil
+	return a.Mpool.SetConfig(cfg)
 }
 
-func (a *MpoolAPI) MpoolSelect(ctx context.Context, tsk types.TipSetKey) ([]*types.SignedMessage, error) {
+func (a *MpoolAPI) MpoolSelect(ctx context.Context, tsk types.TipSetKey, ticketQuality float64) ([]*types.SignedMessage, error) {
 	ts, err := a.Chain.GetTipSetFromKey(tsk)
 	if err != nil {
 		return nil, xerrors.Errorf("loading tipset %s: %w", tsk, err)
 	}
 
-	return a.Mpool.SelectMessages(ts)
+	return a.Mpool.SelectMessages(ts, ticketQuality)
 }
 
 func (a *MpoolAPI) MpoolPending(ctx context.Context, tsk types.TipSetKey) ([]*types.SignedMessage, error) {
@@ -107,67 +106,59 @@ func (a *MpoolAPI) MpoolPending(ctx context.Context, tsk types.TipSetKey) ([]*ty
 	}
 }
 
+func (a *MpoolAPI) MpoolClear(ctx context.Context, local bool) error {
+	a.Mpool.Clear(local)
+	return nil
+}
+
 func (a *MpoolAPI) MpoolPush(ctx context.Context, smsg *types.SignedMessage) (cid.Cid, error) {
 	return a.Mpool.Push(smsg)
 }
 
-func (a *MpoolAPI) MpoolPushMessage(ctx context.Context, msg *types.Message) (*types.SignedMessage, error) {
+func capGasFee(msg *types.Message, maxFee abi.TokenAmount) {
+	if maxFee.Equals(big.Zero()) {
+		return
+	}
+
+	gl := types.NewInt(uint64(msg.GasLimit))
+	totalFee := types.BigMul(msg.GasFeeCap, gl)
+	minerFee := types.BigMul(msg.GasPremium, gl)
+
+	if totalFee.LessThanEqual(maxFee) {
+		return
+	}
+
+	// scale chain/miner fee down proportionally to fit in our budget
+	// TODO: there are probably smarter things we can do here to optimize
+	//  message inclusion latency
+
+	msg.GasFeeCap = big.Div(maxFee, gl)
+	msg.GasPremium = big.Div(big.Div(big.Mul(minerFee, maxFee), totalFee), gl)
+}
+
+func (a *MpoolAPI) MpoolPushMessage(ctx context.Context, msg *types.Message, spec *api.MessageSendSpec) (*types.SignedMessage, error) {
 	{
 		fromA, err := a.Stmgr.ResolveToKeyAddress(ctx, msg.From, nil)
 		if err != nil {
 			return nil, xerrors.Errorf("getting key address: %w", err)
 		}
-
-		a.PushLocks.Lock()
-		if a.PushLocks.m == nil {
-			a.PushLocks.m = make(map[address.Address]chan struct{})
+		done, err := a.PushLocks.TakeLock(ctx, fromA)
+		if err != nil {
+			return nil, xerrors.Errorf("taking lock: %w", err)
 		}
-		lk, ok := a.PushLocks.m[fromA]
-		if !ok {
-			lk = make(chan struct{}, 1)
-			a.PushLocks.m[msg.From] = lk
-		}
-		a.PushLocks.Unlock()
-
-		select {
-		case lk <- struct{}{}:
-		case <-ctx.Done():
-			return nil, ctx.Err()
-		}
-
-		defer func() {
-			<-lk
-		}()
+		defer done()
 	}
 
 	if msg.Nonce != 0 {
 		return nil, xerrors.Errorf("MpoolPushMessage expects message nonce to be 0, was %d", msg.Nonce)
 	}
-	if msg.GasLimit == 0 {
-		gasLimit, err := a.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, err := a.GasAPI.GasEstimateMessageGas(ctx, msg, spec, types.EmptyTSK)
+	if err != nil {
+		return nil, xerrors.Errorf("GasEstimateMessageGas error: %w", err)
 	}
 
-	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{})
-		if err != nil {
-			return nil, xerrors.Errorf("estimating gas price: %w", err)
-		}
-		msg.GasPremium = gasPremium
-	}
-
-	if msg.GasFeeCap == types.EmptyInt || types.BigCmp(msg.GasFeeCap, types.NewInt(0)) == 0 {
-		feeCap, err := a.GasEstimateFeeCap(ctx, msg, 10, types.EmptyTSK)
-		if err != nil {
-			return nil, xerrors.Errorf("estimating fee cap: %w", err)
-		}
-		msg.GasFeeCap = feeCap
-	}
-
-	return a.Mpool.PushWithNonce(ctx, msg.From, func(from address.Address, nonce uint64) (*types.SignedMessage, error) {
+	sign := func(from address.Address, nonce uint64) (*types.SignedMessage, error) {
 		msg.Nonce = nonce
 		if msg.From.Protocol() == address.ID {
 			log.Warnf("Push from ID address (%s), adjusting to %s", msg.From, from)
@@ -184,7 +175,16 @@ func (a *MpoolAPI) MpoolPushMessage(ctx context.Context, msg *types.Message) (*t
 		}
 
 		return a.WalletSignMessage(ctx, from, msg)
-	})
+	}
+
+	var m *types.SignedMessage
+again:
+	m, err = a.Mpool.PushWithNonce(ctx, msg.From, sign)
+	if err == messagepool.ErrTryAgain {
+		log.Debugf("temporary failure while pushing message: %s; retrying", err)
+		goto again
+	}
+	return m, err
 }
 
 func (a *MpoolAPI) MpoolGetNonce(ctx context.Context, addr address.Address) (uint64, error) {

node/impl/full/multisig.go

@@ -28,7 +28,7 @@ type MsigAPI struct {
 	MpoolAPI  MpoolAPI
 }
 
-// TODO: remove gp (gasPrice) from arguemnts
+// TODO: remove gp (gasPrice) from 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) {
 
 	lenAddrs := uint64(len(addrs))
@@ -78,7 +78,7 @@ func (a *MsigAPI) MsigCreate(ctx context.Context, req uint64, addrs []address.Ad
 	}
 
 	// send the message out to the network
-	smsg, err := a.MpoolAPI.MpoolPushMessage(ctx, &msg)
+	smsg, err := a.MpoolAPI.MpoolPushMessage(ctx, &msg, nil)
 	if err != nil {
 		return cid.Undef, err
 	}
@@ -122,7 +122,7 @@ func (a *MsigAPI) MsigPropose(ctx context.Context, msig address.Address, to addr
 		Params: enc,
 	}
 
-	smsg, err := a.MpoolAPI.MpoolPushMessage(ctx, msg)
+	smsg, err := a.MpoolAPI.MpoolPushMessage(ctx, msg, nil)
 	if err != nil {
 		return cid.Undef, xerrors.Errorf("failed to push message: %w", err)
 	}
@@ -236,7 +236,7 @@ func (a *MsigAPI) msigApproveOrCancel(ctx context.Context, operation api.MsigPro
 		Params: enc,
 	}
 
-	smsg, err := a.MpoolAPI.MpoolPushMessage(ctx, msg)
+	smsg, err := a.MpoolAPI.MpoolPushMessage(ctx, msg, nil)
 	if err != nil {
 		return cid.Undef, err
 	}

node/impl/full/state.go

@@ -15,7 +15,7 @@ import (
 
 	"github.com/filecoin-project/go-address"
 	"github.com/filecoin-project/go-bitfield"
-	"github.com/filecoin-project/sector-storage/ffiwrapper"
+	"github.com/filecoin-project/lotus/extern/sector-storage/ffiwrapper"
 	"github.com/filecoin-project/specs-actors/actors/abi"
 	"github.com/filecoin-project/specs-actors/actors/abi/big"
 	"github.com/filecoin-project/specs-actors/actors/builtin"
@@ -74,7 +74,7 @@ func (a *StateAPI) StateMinerActiveSectors(ctx context.Context, maddr address.Ad
 	err := a.StateManager.WithParentStateTsk(tsk,
 		a.StateManager.WithActor(maddr,
 			a.StateManager.WithActorState(ctx, func(store adt.Store, mas *miner.State) error {
-				var allActive []*abi.BitField
+				var allActive []abi.BitField
 
 				err := a.StateManager.WithDeadlines(
 					a.StateManager.WithEachDeadline(
@@ -96,7 +96,7 @@ func (a *StateAPI) StateMinerActiveSectors(ctx context.Context, maddr address.Ad
 					return xerrors.Errorf("merging active sector bitfields: %w", err)
 				}
 
-				out, err = stmgr.LoadSectorsFromSet(ctx, a.Chain.Blockstore(), mas.Sectors, active, false)
+				out, err = stmgr.LoadSectorsFromSet(ctx, a.Chain.Blockstore(), mas.Sectors, &active, false)
 				return err
 			})))
 	if err != nil {
@@ -160,8 +160,8 @@ func (a *StateAPI) StateMinerProvingDeadline(ctx context.Context, addr address.A
 	return mas.DeadlineInfo(ts.Height()).NextNotElapsed(), nil
 }
 
-func (a *StateAPI) StateMinerFaults(ctx context.Context, addr address.Address, tsk types.TipSetKey) (*abi.BitField, error) {
-	out := abi.NewBitField()
+func (a *StateAPI) StateMinerFaults(ctx context.Context, addr address.Address, tsk types.TipSetKey) (abi.BitField, error) {
+	out := bitfield.New()
 
 	err := a.StateManager.WithParentStateTsk(tsk,
 		a.StateManager.WithActor(addr,
@@ -173,7 +173,7 @@ func (a *StateAPI) StateMinerFaults(ctx context.Context, addr address.Address, t
 							return err
 						}))))))
 	if err != nil {
-		return nil, err
+		return bitfield.BitField{}, err
 	}
 
 	return out, err
@@ -203,7 +203,7 @@ func (a *StateAPI) StateAllMinerFaults(ctx context.Context, lookback abi.ChainEp
 			return nil, xerrors.Errorf("failed to load miner actor state %s: %w", m, err)
 		}
 
-		err = mas.ForEachFaultEpoch(a.Chain.Store(ctx), func(faultStart abi.ChainEpoch, faults *abi.BitField) error {
+		err = mas.ForEachFaultEpoch(a.Chain.Store(ctx), func(faultStart abi.ChainEpoch, faults abi.BitField) error {
 			if faultStart >= cutoff {
 				allFaults = append(allFaults, &api.Fault{
 					Miner: m,
@@ -222,8 +222,8 @@ func (a *StateAPI) StateAllMinerFaults(ctx context.Context, lookback abi.ChainEp
 	return allFaults, nil*/
 }
 
-func (a *StateAPI) StateMinerRecoveries(ctx context.Context, addr address.Address, tsk types.TipSetKey) (*abi.BitField, error) {
-	out := abi.NewBitField()
+func (a *StateAPI) StateMinerRecoveries(ctx context.Context, addr address.Address, tsk types.TipSetKey) (abi.BitField, error) {
+	out := bitfield.New()
 
 	err := a.StateManager.WithParentStateTsk(tsk,
 		a.StateManager.WithActor(addr,
@@ -235,7 +235,7 @@ func (a *StateAPI) StateMinerRecoveries(ctx context.Context, addr address.Addres
 							return err
 						}))))))
 	if err != nil {
-		return nil, err
+		return bitfield.BitField{}, err
 	}
 
 	return out, err
@@ -258,37 +258,6 @@ func (a *StateAPI) StateMinerPower(ctx context.Context, addr address.Address, ts
 	}, nil
 }
 
-func (a *StateAPI) StatePledgeCollateral(ctx context.Context, tsk types.TipSetKey) (types.BigInt, error) {
-	/*ts, err := a.Chain.GetTipSetFromKey(tsk)
-	if err != nil {
-		return types.EmptyInt, xerrors.Errorf("loading tipset %s: %w", tsk, err)
-	}
-
-	param, err := actors.SerializeParams(&actors.PledgeCollateralParams{Size: types.NewInt(0)})
-	if err != nil {
-		return types.NewInt(0), err
-	}
-
-	ret, aerr := a.StateManager.Call(ctx, &types.Message{
-		From:   actors.StoragePowerAddress,
-		To:     actors.StoragePowerAddress,
-		Method: actors.SPAMethods.PledgeCollateralForSize,
-
-		Params: param,
-	}, ts)
-	if aerr != nil {
-		return types.NewInt(0), xerrors.Errorf("failed to get miner worker addr: %w", err)
-	}
-
-	if ret.MsgRct.ExitCode != 0 {
-		return types.NewInt(0), xerrors.Errorf("failed to get miner worker addr (exit code %d)", ret.MsgRct.ExitCode)
-	}
-
-	return types.BigFromBytes(ret.Return), nil*/
-	log.Error("TODO StatePledgeCollateral")
-	return big.Zero(), nil
-}
-
 func (a *StateAPI) StateCall(ctx context.Context, msg *types.Message, tsk types.TipSetKey) (*api.InvocResult, error) {
 	ts, err := a.Chain.GetTipSetFromKey(tsk)
 	if err != nil {
@@ -426,7 +395,7 @@ func (a *StateAPI) MinerCreateBlock(ctx context.Context, bt *api.BlockTemplate)
 }
 
 func (a *StateAPI) StateWaitMsg(ctx context.Context, msg cid.Cid, confidence uint64) (*api.MsgLookup, error) {
-	ts, recpt, err := a.StateManager.WaitForMessage(ctx, msg, confidence)
+	ts, recpt, found, err := a.StateManager.WaitForMessage(ctx, msg, confidence)
 	if err != nil {
 		return nil, err
 	}
@@ -453,6 +422,7 @@ func (a *StateAPI) StateWaitMsg(ctx context.Context, msg cid.Cid, confidence uin
 	}
 
 	return &api.MsgLookup{
+		Message:   found,
 		Receipt:   *recpt,
 		ReturnDec: returndec,
 		TipSet:    ts.Key(),
@@ -461,20 +431,20 @@ func (a *StateAPI) StateWaitMsg(ctx context.Context, msg cid.Cid, confidence uin
 }
 
 func (a *StateAPI) StateSearchMsg(ctx context.Context, msg cid.Cid) (*api.MsgLookup, error) {
-	ts, recpt, err := a.StateManager.SearchForMessage(ctx, msg)
+	ts, recpt, found, err := a.StateManager.SearchForMessage(ctx, msg)
 	if err != nil {
 		return nil, err
 	}
 
 	if ts != nil {
 		return &api.MsgLookup{
+			Message: found,
 			Receipt: *recpt,
 			TipSet:  ts.Key(),
 			Height:  ts.Height(),
 		}, nil
-	} else {
-		return nil, nil
 	}
+	return nil, nil
 }
 
 func (a *StateAPI) StateGetReceipt(ctx context.Context, msg cid.Cid, tsk types.TipSetKey) (*types.MessageReceipt, error) {
@@ -586,7 +556,7 @@ func (a *StateAPI) StateMarketDeals(ctx context.Context, tsk types.TipSetKey) (m
 		} else if !found {
 			s.SectorStartEpoch = -1
 		}
-		out[strconv.FormatInt(int64(i), 10)] = api.MarketDeal{
+		out[strconv.FormatInt(i, 10)] = api.MarketDeal{
 			Proposal: d,
 			State:    s,
 		}
@@ -666,7 +636,7 @@ func (a *StateAPI) StateMinerSectorCount(ctx context.Context, addr address.Addre
 	err := a.StateManager.WithParentStateTsk(tsk,
 		a.StateManager.WithActor(addr,
 			a.StateManager.WithActorState(ctx, func(store adt.Store, mas *miner.State) error {
-				var allActive []*abi.BitField
+				var allActive []abi.BitField
 
 				err := a.StateManager.WithDeadlines(
 					a.StateManager.WithEachDeadline(
@@ -814,10 +784,6 @@ func (a *StateAPI) StateSectorPartition(ctx context.Context, maddr address.Addre
 		return nil, err
 	}
 
-	if found == nil {
-
-	}
-
 	return found, nil
 }
 
@@ -1069,7 +1035,7 @@ func (a *StateAPI) StateMinerInitialPledgeCollateral(ctx context.Context, maddr
 
 	duration := pci.Expiration - ts.Height() // NB: not exactly accurate, but should always lead us to *over* estimate, not under
 
-	circSupply, err := a.StateManager.CirculatingSupply(ctx, ts)
+	circSupply, err := a.StateCirculatingSupply(ctx, ts.Key())
 	if err != nil {
 		return big.Zero(), xerrors.Errorf("getting circulating supply: %w", err)
 	}
@@ -1081,7 +1047,7 @@ func (a *StateAPI) StateMinerInitialPledgeCollateral(ctx context.Context, maddr
 		powerState.ThisEpochPledgeCollateral,
 		rewardState.ThisEpochRewardSmoothed,
 		powerState.ThisEpochQAPowerSmoothed,
-		circSupply,
+		circSupply.FilCirculating,
 	)
 
 	return types.BigDiv(types.BigMul(initialPledge, initialPledgeNum), initialPledgeDen), nil
@@ -1121,6 +1087,12 @@ func (a *StateAPI) StateVerifiedClientStatus(ctx context.Context, addr address.A
 		return nil, err
 	}
 
+	aid, err := a.StateLookupID(ctx, addr, tsk)
+	if err != nil {
+		log.Warnf("lookup failure %v", err)
+		return nil, err
+	}
+
 	store := a.StateManager.ChainStore().Store(ctx)
 
 	var st verifreg.State
@@ -1134,7 +1106,7 @@ func (a *StateAPI) StateVerifiedClientStatus(ctx context.Context, addr address.A
 	}
 
 	var dcap verifreg.DataCap
-	if found, err := vh.Get(adt.AddrKey(addr), &dcap); err != nil {
+	if found, err := vh.Get(adt.AddrKey(aid), &dcap); err != nil {
 		return nil, err
 	} else if !found {
 		return nil, nil
@@ -1170,26 +1142,37 @@ func (a *StateAPI) StateDealProviderCollateralBounds(ctx context.Context, size a
 	})
 
 	if err != nil {
-		return api.DealCollateralBounds{}, xerrors.Errorf("getting power and reward actor states: %w")
+		return api.DealCollateralBounds{}, xerrors.Errorf("getting power and reward actor states: %w", err)
 	}
 
-	circ, err := a.StateManager.CirculatingSupply(ctx, ts)
+	circ, err := a.StateCirculatingSupply(ctx, ts.Key())
 	if err != nil {
-		return api.DealCollateralBounds{}, xerrors.Errorf("getting total circulating supply: %w")
+		return api.DealCollateralBounds{}, xerrors.Errorf("getting total circulating supply: %w", err)
 	}
 
-	min, max := market.DealProviderCollateralBounds(size, verified, powerState.ThisEpochQualityAdjPower, rewardState.ThisEpochBaselinePower, circ)
+	min, max := market.DealProviderCollateralBounds(size, verified, powerState.ThisEpochQualityAdjPower, rewardState.ThisEpochBaselinePower, circ.FilCirculating)
 	return api.DealCollateralBounds{
 		Min: types.BigDiv(types.BigMul(min, dealProviderCollateralNum), dealProviderCollateralDen),
 		Max: max,
 	}, nil
 }
 
-func (a *StateAPI) StateCirculatingSupply(ctx context.Context, tsk types.TipSetKey) (abi.TokenAmount, error) {
+func (a *StateAPI) StateCirculatingSupply(ctx context.Context, tsk types.TipSetKey) (api.CirculatingSupply, error) {
 	ts, err := a.Chain.GetTipSetFromKey(tsk)
 	if err != nil {
-		return abi.TokenAmount{}, xerrors.Errorf("loading tipset %s: %w", tsk, err)
+		return api.CirculatingSupply{}, xerrors.Errorf("loading tipset %s: %w", tsk, err)
 	}
 
-	return stmgr.GetCirculatingSupply(ctx, a.StateManager, ts)
+	st, _, err := a.StateManager.TipSetState(ctx, ts)
+	if err != nil {
+		return api.CirculatingSupply{}, err
+	}
+
+	cst := cbor.NewCborStore(a.Chain.Blockstore())
+	sTree, err := state.LoadStateTree(cst, st)
+	if err != nil {
+		return api.CirculatingSupply{}, err
+	}
+
+	return a.StateManager.GetCirculatingSupplyDetailed(ctx, ts.Height(), sTree)
 }

node/impl/full/sync.go

@@ -90,8 +90,7 @@ func (a *SyncAPI) SyncSubmitBlock(ctx context.Context, blk *types.BlockMsg) erro
 		return xerrors.Errorf("serializing block for pubsub publishing failed: %w", err)
 	}
 
-	// TODO: anything else to do here?
-	return a.PubSub.Publish(build.BlocksTopic(a.NetName), b)
+	return a.PubSub.Publish(build.BlocksTopic(a.NetName), b) //nolint:staticcheck
 }
 
 func (a *SyncAPI) SyncIncomingBlocks(ctx context.Context) (<-chan *types.BlockHeader, error) {

node/impl/full/wallet.go

@@ -2,20 +2,18 @@ package full
 
 import (
 	"context"
-	"github.com/filecoin-project/specs-actors/actors/abi/big"
-	init_ "github.com/filecoin-project/specs-actors/actors/builtin/init"
-
-	"github.com/filecoin-project/lotus/lib/sigs"
 
 	"github.com/filecoin-project/go-address"
+	"go.uber.org/fx"
+	"golang.org/x/xerrors"
+
+	"github.com/filecoin-project/specs-actors/actors/abi/big"
 	"github.com/filecoin-project/specs-actors/actors/crypto"
 
 	"github.com/filecoin-project/lotus/chain/stmgr"
 	"github.com/filecoin-project/lotus/chain/types"
 	"github.com/filecoin-project/lotus/chain/wallet"
-
-	"go.uber.org/fx"
-	"golang.org/x/xerrors"
+	"github.com/filecoin-project/lotus/lib/sigs"
 )
 
 type WalletAPI struct {
@@ -44,11 +42,10 @@ func (a *WalletAPI) WalletBalance(ctx context.Context, addr address.Address) (ty
 		return nil
 	}))
 
-	if xerrors.Is(err, init_.ErrAddressNotFound) {
+	if xerrors.Is(err, types.ErrActorNotFound) {
 		return big.Zero(), nil
-	} else {
-		return bal, err
 	}
+	return bal, err
 }
 
 func (a *WalletAPI) WalletSign(ctx context.Context, k address.Address, msg []byte) (*crypto.Signature, error) {

node/impl/paych/paych.go

@@ -4,13 +4,14 @@ import (
 	"context"
 	"fmt"
 
-	"github.com/filecoin-project/specs-actors/actors/builtin"
-	"github.com/filecoin-project/specs-actors/actors/builtin/paych"
 	"github.com/ipfs/go-cid"
 	"go.uber.org/fx"
 	"golang.org/x/xerrors"
 
 	"github.com/filecoin-project/go-address"
+	"github.com/filecoin-project/specs-actors/actors/builtin"
+	"github.com/filecoin-project/specs-actors/actors/builtin/paych"
+
 	"github.com/filecoin-project/lotus/api"
 	"github.com/filecoin-project/lotus/chain/actors"
 	"github.com/filecoin-project/lotus/chain/types"
@@ -35,13 +36,13 @@ func (a *PaychAPI) PaychGet(ctx context.Context, from, to address.Address, amt t
 	}
 
 	return &api.ChannelInfo{
-		Channel:        ch,
-		ChannelMessage: mcid,
+		Channel:      ch,
+		WaitSentinel: mcid,
 	}, nil
 }
 
-func (a *PaychAPI) PaychGetWaitReady(ctx context.Context, mcid cid.Cid) (address.Address, error) {
-	return a.PaychMgr.GetPaychWaitReady(ctx, mcid)
+func (a *PaychAPI) PaychGetWaitReady(ctx context.Context, sentinel cid.Cid) (address.Address, error) {
+	return a.PaychMgr.GetPaychWaitReady(ctx, sentinel)
 }
 
 func (a *PaychAPI) PaychAllocateLane(ctx context.Context, ch address.Address) (uint64, error) {
@@ -84,15 +85,10 @@ func (a *PaychAPI) PaychNewPayment(ctx context.Context, from, to address.Address
 		svs[i] = sv
 	}
 
-	var pchCid *cid.Cid
-	if ch.ChannelMessage != cid.Undef {
-		pchCid = &ch.ChannelMessage
-	}
-
 	return &api.PaymentInfo{
-		Channel:        ch.Channel,
-		ChannelMessage: pchCid,
-		Vouchers:       svs,
+		Channel:      ch.Channel,
+		WaitSentinel: ch.WaitSentinel,
+		Vouchers:     svs,
 	}, nil
 }
 
@@ -128,9 +124,7 @@ func (a *PaychAPI) PaychVoucherCheckSpendable(ctx context.Context, ch address.Ad
 }
 
 func (a *PaychAPI) PaychVoucherAdd(ctx context.Context, ch address.Address, sv *paych.SignedVoucher, proof []byte, minDelta types.BigInt) (types.BigInt, error) {
-	_ = a.PaychMgr.TrackInboundChannel(ctx, ch) // TODO: expose those calls
-
-	return a.PaychMgr.AddVoucher(ctx, ch, sv, proof, minDelta)
+	return a.PaychMgr.AddVoucherInbound(ctx, ch, sv, proof, minDelta)
 }
 
 // PaychVoucherCreate creates a new signed voucher on the given payment channel
@@ -168,7 +162,7 @@ func (a *PaychAPI) paychVoucherCreate(ctx context.Context, pch address.Address,
 
 	sv.Signature = sig
 
-	if _, err := a.PaychMgr.AddVoucher(ctx, pch, sv, nil, types.NewInt(0)); err != nil {
+	if _, err := a.PaychMgr.AddVoucherOutbound(ctx, pch, sv, nil, types.NewInt(0)); err != nil {
 		return nil, xerrors.Errorf("failed to persist voucher: %w", err)
 	}
 
@@ -214,7 +208,7 @@ func (a *PaychAPI) PaychVoucherSubmit(ctx context.Context, ch address.Address, s
 		Params: enc,
 	}
 
-	smsg, err := a.MpoolPushMessage(ctx, msg)
+	smsg, err := a.MpoolPushMessage(ctx, msg, nil)
 	if err != nil {
 		return cid.Undef, err
 	}

node/impl/remoteworker.go

@@ -9,11 +9,10 @@ import (
 	"github.com/filecoin-project/go-jsonrpc"
 	"github.com/filecoin-project/go-jsonrpc/auth"
 	"github.com/filecoin-project/specs-actors/actors/abi"
-	storage2 "github.com/filecoin-project/specs-storage/storage"
 
 	"github.com/filecoin-project/lotus/api"
 	"github.com/filecoin-project/lotus/api/client"
-	"github.com/filecoin-project/sector-storage"
+	sectorstorage "github.com/filecoin-project/lotus/extern/sector-storage"
 )
 
 type remoteWorker struct {
@@ -25,10 +24,6 @@ func (r *remoteWorker) NewSector(ctx context.Context, sector abi.SectorID) error
 	return xerrors.New("unsupported")
 }
 
-func (r *remoteWorker) AddPiece(ctx context.Context, sector abi.SectorID, pieceSizes []abi.UnpaddedPieceSize, newPieceSize abi.UnpaddedPieceSize, pieceData storage2.Data) (abi.PieceInfo, error) {
-	return abi.PieceInfo{}, xerrors.New("unsupported")
-}
-
 func connectRemoteWorker(ctx context.Context, fa api.Common, url string) (*remoteWorker, error) {
 	token, err := fa.AuthNew(ctx, []auth.Permission{"admin"})
 	if err != nil {
@@ -38,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(url, headers)
+	wapi, closer, err := client.NewWorkerRPC(context.TODO(), url, headers)
 	if err != nil {
 		return nil, xerrors.Errorf("creating jsonrpc client: %w", err)
 	}

node/impl/storminer.go

@@ -8,9 +8,10 @@ import (
 	"strconv"
 	"time"
 
-	"github.com/filecoin-project/sector-storage/fsutil"
+	datatransfer "github.com/filecoin-project/go-data-transfer"
 	"github.com/filecoin-project/specs-actors/actors/abi/big"
 	"github.com/ipfs/go-cid"
+	"github.com/libp2p/go-libp2p-core/host"
 	"golang.org/x/xerrors"
 
 	"github.com/filecoin-project/go-address"
@@ -18,12 +19,14 @@ import (
 	retrievalmarket "github.com/filecoin-project/go-fil-markets/retrievalmarket"
 	storagemarket "github.com/filecoin-project/go-fil-markets/storagemarket"
 	"github.com/filecoin-project/go-jsonrpc/auth"
-	sectorstorage "github.com/filecoin-project/sector-storage"
-	"github.com/filecoin-project/sector-storage/ffiwrapper"
-	"github.com/filecoin-project/sector-storage/stores"
-	"github.com/filecoin-project/sector-storage/storiface"
 	"github.com/filecoin-project/specs-actors/actors/abi"
-	sealing "github.com/filecoin-project/storage-fsm"
+
+	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/fsutil"
+	"github.com/filecoin-project/lotus/extern/sector-storage/stores"
+	"github.com/filecoin-project/lotus/extern/sector-storage/storiface"
+	sealing "github.com/filecoin-project/lotus/extern/storage-sealing"
 
 	"github.com/filecoin-project/lotus/api"
 	"github.com/filecoin-project/lotus/api/apistruct"
@@ -50,6 +53,8 @@ type StorageMinerAPI struct {
 	StorageMgr        *sectorstorage.Manager `optional:"true"`
 	IStorageMgr       sectorstorage.SectorManager
 	*stores.Index
+	DataTransfer dtypes.ProviderDataTransfer
+	Host         host.Host
 
 	ConsiderOnlineStorageDealsConfigFunc       dtypes.ConsiderOnlineStorageDealsConfigFunc
 	SetConsiderOnlineStorageDealsConfigFunc    dtypes.SetConsiderOnlineStorageDealsConfigFunc
@@ -61,8 +66,8 @@ type StorageMinerAPI struct {
 	SetConsiderOfflineStorageDealsConfigFunc   dtypes.SetConsiderOfflineStorageDealsConfigFunc
 	ConsiderOfflineRetrievalDealsConfigFunc    dtypes.ConsiderOfflineRetrievalDealsConfigFunc
 	SetConsiderOfflineRetrievalDealsConfigFunc dtypes.SetConsiderOfflineRetrievalDealsConfigFunc
-	SetSealingDelayFunc                        dtypes.SetSealingDelayFunc
-	GetSealingDelayFunc                        dtypes.GetSealingDelayFunc
+	SetSealingConfigFunc                       dtypes.SetSealingConfigFunc
+	GetSealingConfigFunc                       dtypes.GetSealingConfigFunc
 	GetExpectedSealDurationFunc                dtypes.GetExpectedSealDurationFunc
 	SetExpectedSealDurationFunc                dtypes.SetExpectedSealDurationFunc
 }
@@ -148,7 +153,8 @@ func (sm *StorageMinerAPI) SectorsStatus(ctx context.Context, sid abi.SectorNumb
 			Value: info.SeedValue,
 			Epoch: info.SeedEpoch,
 		},
-		Retries: info.InvalidProofs,
+		Retries:   info.InvalidProofs,
+		ToUpgrade: sm.Miner.IsMarkedForUpgrade(sid),
 
 		LastErr: info.LastErr,
 		Log:     log,
@@ -231,11 +237,22 @@ func (sm *StorageMinerAPI) SectorStartSealing(ctx context.Context, number abi.Se
 }
 
 func (sm *StorageMinerAPI) SectorSetSealDelay(ctx context.Context, delay time.Duration) error {
-	return sm.SetSealingDelayFunc(delay)
+	cfg, err := sm.GetSealingConfigFunc()
+	if err != nil {
+		return xerrors.Errorf("get config: %w", err)
+	}
+
+	cfg.WaitDealsDelay = delay
+
+	return sm.SetSealingConfigFunc(cfg)
 }
 
 func (sm *StorageMinerAPI) SectorGetSealDelay(ctx context.Context) (time.Duration, error) {
-	return sm.GetSealingDelayFunc()
+	cfg, err := sm.GetSealingConfigFunc()
+	if err != nil {
+		return 0, err
+	}
+	return cfg.WaitDealsDelay, nil
 }
 
 func (sm *StorageMinerAPI) SectorSetExpectedSealDuration(ctx context.Context, delay time.Duration) error {
@@ -342,6 +359,39 @@ func (sm *StorageMinerAPI) MarketGetRetrievalAsk(ctx context.Context) (*retrieva
 	return sm.RetrievalProvider.GetAsk(), nil
 }
 
+func (sm *StorageMinerAPI) MarketListDataTransfers(ctx context.Context) ([]api.DataTransferChannel, error) {
+	inProgressChannels, err := sm.DataTransfer.InProgressChannels(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	apiChannels := make([]api.DataTransferChannel, 0, len(inProgressChannels))
+	for _, channelState := range inProgressChannels {
+		apiChannels = append(apiChannels, api.NewDataTransferChannel(sm.Host.ID(), channelState))
+	}
+
+	return apiChannels, nil
+}
+
+func (sm *StorageMinerAPI) MarketDataTransferUpdates(ctx context.Context) (<-chan api.DataTransferChannel, error) {
+	channels := make(chan api.DataTransferChannel)
+
+	unsub := sm.DataTransfer.SubscribeToEvents(func(evt datatransfer.Event, channelState datatransfer.ChannelState) {
+		channel := api.NewDataTransferChannel(sm.Host.ID(), channelState)
+		select {
+		case <-ctx.Done():
+		case channels <- channel:
+		}
+	})
+
+	go func() {
+		defer unsub()
+		<-ctx.Done()
+	}()
+
+	return channels, nil
+}
+
 func (sm *StorageMinerAPI) DealsList(ctx context.Context) ([]storagemarket.StorageDeal, error) {
 	return sm.StorageProvider.ListDeals(ctx)
 }

node/modules/chain.go

@@ -3,6 +3,7 @@ package modules
 import (
 	"bytes"
 	"context"
+	"os"
 
 	"github.com/ipfs/go-bitswap"
 	"github.com/ipfs/go-bitswap/network"
@@ -15,7 +16,7 @@ import (
 	"go.uber.org/fx"
 	"golang.org/x/xerrors"
 
-	"github.com/filecoin-project/sector-storage/ffiwrapper"
+	"github.com/filecoin-project/lotus/extern/sector-storage/ffiwrapper"
 
 	"github.com/filecoin-project/lotus/chain"
 	"github.com/filecoin-project/lotus/chain/beacon"
@@ -127,8 +128,18 @@ func LoadGenesis(genBytes []byte) func(dtypes.ChainBlockstore) Genesis {
 func DoSetGenesis(_ dtypes.AfterGenesisSet) {}
 
 func SetGenesis(cs *store.ChainStore, g Genesis) (dtypes.AfterGenesisSet, error) {
-	_, err := cs.GetGenesis()
+	genFromRepo, err := cs.GetGenesis()
 	if err == nil {
+		if os.Getenv("LOTUS_SKIP_GENESIS_CHECK") != "_yes_" {
+			expectedGenesis, err := g()
+			if err != nil {
+				return dtypes.AfterGenesisSet{}, xerrors.Errorf("getting expected genesis failed: %w", err)
+			}
+
+			if genFromRepo.Cid() != expectedGenesis.Cid() {
+				return dtypes.AfterGenesisSet{}, xerrors.Errorf("genesis in the repo is not the one expected by this version of Lotus!")
+			}
+		}
 		return dtypes.AfterGenesisSet{}, nil // already set, noop
 	}
 	if err != datastore.ErrNotFound {

node/modules/client.go

@@ -21,7 +21,6 @@ import (
 	"github.com/filecoin-project/go-fil-markets/storagemarket/impl/funds"
 	"github.com/filecoin-project/go-fil-markets/storagemarket/impl/requestvalidation"
 	smnet "github.com/filecoin-project/go-fil-markets/storagemarket/network"
-	"github.com/filecoin-project/go-statestore"
 	"github.com/filecoin-project/go-storedcounter"
 	"github.com/ipfs/go-datastore"
 	"github.com/ipfs/go-datastore/namespace"
@@ -101,20 +100,11 @@ func NewClientGraphsyncDataTransfer(lc fx.Lifecycle, h host.Host, gs dtypes.Grap
 	return dt, nil
 }
 
-// NewClientDealStore creates a statestore for the client to store its deals
-func NewClientDealStore(ds dtypes.ClientDatastore) dtypes.ClientDealStore {
-	return statestore.New(ds)
-}
-
 // NewClientDatastore creates a datastore for the client to store its deals
 func NewClientDatastore(ds dtypes.MetadataDS) dtypes.ClientDatastore {
 	return namespace.Wrap(ds, datastore.NewKey("/deals/client"))
 }
 
-func NewClientRequestValidator(deals dtypes.ClientDealStore) dtypes.ClientRequestValidator {
-	return requestvalidation.NewUnifiedRequestValidator(nil, deals)
-}
-
 type ClientDealFunds funds.DealFunds
 
 func NewClientDealFunds(ds dtypes.MetadataDS) (ClientDealFunds, error) {
@@ -133,8 +123,7 @@ func StorageClient(lc fx.Lifecycle, h host.Host, ibs dtypes.ClientBlockstore, md
 			return c.Start(ctx)
 		},
 		OnStop: func(context.Context) error {
-			c.Stop()
-			return nil
+			return c.Stop()
 		},
 	})
 	return c, nil

node/modules/core.go

@@ -34,9 +34,10 @@ func RecordValidator(ps peerstore.Peerstore) record.Validator {
 	}
 }
 
-const JWTSecretName = "auth-jwt-private" //nolint:gosec
+const JWTSecretName = "auth-jwt-private"  //nolint:gosec
+const KTJwtHmacSecret = "jwt-hmac-secret" //nolint:gosec
 
-type jwtPayload struct {
+type JwtPayload struct {
 	Allow []auth.Permission
 }
 
@@ -52,7 +53,7 @@ func APISecret(keystore types.KeyStore, lr repo.LockedRepo) (*dtypes.APIAlg, err
 		}
 
 		key = types.KeyInfo{
-			Type:       "jwt-hmac-secret",
+			Type:       KTJwtHmacSecret,
 			PrivateKey: sk,
 		}
 
@@ -61,7 +62,7 @@ func APISecret(keystore types.KeyStore, lr repo.LockedRepo) (*dtypes.APIAlg, err
 		}
 
 		// TODO: make this configurable
-		p := jwtPayload{
+		p := JwtPayload{
 			Allow: apistruct.AllPermissions,
 		}
 

node/modules/dtypes/miner.go

@@ -9,6 +9,8 @@ import (
 	"github.com/filecoin-project/go-address"
 	"github.com/filecoin-project/go-fil-markets/storagemarket"
 	"github.com/filecoin-project/specs-actors/actors/abi"
+
+	"github.com/filecoin-project/lotus/extern/storage-sealing/sealiface"
 )
 
 type MinerAddress address.Address
@@ -56,10 +58,10 @@ type ConsiderOfflineRetrievalDealsConfigFunc func() (bool, error)
 type SetConsiderOfflineRetrievalDealsConfigFunc func(bool) error
 
 // SetSealingDelay sets how long a sector waits for more deals before sealing begins.
-type SetSealingDelayFunc func(time.Duration) error
+type SetSealingConfigFunc func(sealiface.Config) error
 
 // GetSealingDelay returns how long a sector waits for more deals before sealing begins.
-type GetSealingDelayFunc func() (time.Duration, error)
+type GetSealingConfigFunc func() (sealiface.Config, error)
 
 // SetExpectedSealDurationFunc is a function which is used to set how long sealing is expected to take.
 // Deals that would need to start earlier than this duration will be rejected.

node/modules/dtypes/mpool.go

@@ -0,0 +1,35 @@
+package dtypes
+
+import (
+	"context"
+	"sync"
+
+	"github.com/filecoin-project/go-address"
+)
+
+type MpoolLocker struct {
+	m  map[address.Address]chan struct{}
+	lk sync.Mutex
+}
+
+func (ml *MpoolLocker) TakeLock(ctx context.Context, a address.Address) (func(), error) {
+	ml.lk.Lock()
+	if ml.m == nil {
+		ml.m = make(map[address.Address]chan struct{})
+	}
+	lk, ok := ml.m[a]
+	if !ok {
+		lk = make(chan struct{}, 1)
+		ml.m[a] = lk
+	}
+	ml.lk.Unlock()
+
+	select {
+	case lk <- struct{}{}:
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+	return func() {
+		<-lk
+	}, nil
+}

node/modules/graphsync.go

@@ -8,7 +8,7 @@ import (
 	gsnet "github.com/ipfs/go-graphsync/network"
 	"github.com/ipfs/go-graphsync/storeutil"
 	"github.com/libp2p/go-libp2p-core/host"
-	peer "github.com/libp2p/go-libp2p-peer"
+	"github.com/libp2p/go-libp2p-core/peer"
 	"go.uber.org/fx"
 )
 

node/modules/lp2p/pubsub.go

@@ -11,10 +11,12 @@ import (
 	pubsub_pb "github.com/libp2p/go-libp2p-pubsub/pb"
 	blake2b "github.com/minio/blake2b-simd"
 	ma "github.com/multiformats/go-multiaddr"
+	"go.opencensus.io/stats"
 	"go.uber.org/fx"
 	"golang.org/x/xerrors"
 
 	"github.com/filecoin-project/lotus/build"
+	"github.com/filecoin-project/lotus/metrics"
 	"github.com/filecoin-project/lotus/node/config"
 	"github.com/filecoin-project/lotus/node/modules/dtypes"
 	"github.com/filecoin-project/lotus/node/modules/helpers"
@@ -29,6 +31,8 @@ func init() {
 	pubsub.GossipSubDhi = 12
 	pubsub.GossipSubDlazy = 12
 	pubsub.GossipSubDirectConnectInitialDelay = 30 * time.Second
+	pubsub.GossipSubIWantFollowupTime = 5 * time.Second
+	pubsub.GossipSubHistoryLength = 10
 }
 func ScoreKeeper() *dtypes.ScoreKeeper {
 	return new(dtypes.ScoreKeeper)
@@ -108,8 +112,9 @@ func GossipSub(in GossipIn) (service *pubsub.PubSub, err error) {
 				// IPColocationFactorWhitelist: map[string]struct{}{},
 
 				// P7: behavioural penalties, decay after 1hr
-				BehaviourPenaltyWeight: -10,
-				BehaviourPenaltyDecay:  pubsub.ScoreParameterDecay(time.Hour),
+				BehaviourPenaltyThreshold: 6,
+				BehaviourPenaltyWeight:    -10,
+				BehaviourPenaltyDecay:     pubsub.ScoreParameterDecay(time.Hour),
 
 				DecayInterval: pubsub.DefaultDecayInterval,
 				DecayToZero:   pubsub.DefaultDecayToZero,
@@ -128,12 +133,12 @@ func GossipSub(in GossipIn) (service *pubsub.PubSub, err error) {
 						TimeInMeshQuantum: time.Second,
 						TimeInMeshCap:     1,
 
-						// deliveries decay after 1 hour, cap at 100 blocks
-						FirstMessageDeliveriesWeight: 5, // max value is 500
+						// deliveries decay after 1 hour, cap at 25 beacons
+						FirstMessageDeliveriesWeight: 5, // max value is 125
 						FirstMessageDeliveriesDecay:  pubsub.ScoreParameterDecay(time.Hour),
-						FirstMessageDeliveriesCap:    100, // 100 blocks in an hour
+						FirstMessageDeliveriesCap:    25, // the maximum expected in an hour is ~26, including the decay
 
-						// Mesh Delivery Failure is currently turned off for blocks
+						// Mesh Delivery Failure is currently turned off for beacons
 						// This is on purpose as
 						// - the traffic is very low for meaningful distribution of incoming edges.
 						// - the reaction time needs to be very slow -- in the order of 10 min at least
@@ -142,19 +147,6 @@ func GossipSub(in GossipIn) (service *pubsub.PubSub, err error) {
 						// - the network is too small, so large asymmetries can be expected between mesh
 						//   edges.
 						// We should revisit this once the network grows.
-						//
-						// // tracks deliveries in the last minute
-						// // penalty activates at 1 minute and expects ~0.4 blocks
-						// MeshMessageDeliveriesWeight:     -576, // max penalty is -100
-						// MeshMessageDeliveriesDecay:      pubsub.ScoreParameterDecay(time.Minute),
-						// MeshMessageDeliveriesCap:        10,      // 10 blocks in a minute
-						// MeshMessageDeliveriesThreshold:  0.41666, // 10/12/2 blocks/min
-						// MeshMessageDeliveriesWindow:     10 * time.Millisecond,
-						// MeshMessageDeliveriesActivation: time.Minute,
-						//
-						// // decays after 15 min
-						// MeshFailurePenaltyWeight: -576,
-						// MeshFailurePenaltyDecay:  pubsub.ScoreParameterDecay(15 * time.Minute),
 
 						// invalid messages decay after 1 hour
 						InvalidMessageDeliveriesWeight: -1000,
@@ -303,12 +295,12 @@ func GossipSub(in GossipIn) (service *pubsub.PubSub, err error) {
 
 		trw := newTracerWrapper(tr, build.BlocksTopic(in.Nn))
 		options = append(options, pubsub.WithEventTracer(trw))
+	} else {
+		// still instantiate a tracer for collecting metrics
+		trw := newTracerWrapper(nil)
+		options = append(options, pubsub.WithEventTracer(trw))
 	}
 
-	// TODO: we want to hook the peer score inspector so that we can gain visibility
-	//       in peer scores for debugging purposes -- this might be trigged by metrics collection
-	// options = append(options, pubsub.WithPeerScoreInspect(XXX, time.Second))
-
 	return pubsub.NewGossipSub(helpers.LifecycleCtx(in.Mctx, in.Lc), in.Host, options...)
 }
 
@@ -318,10 +310,14 @@ func HashMsgId(m *pubsub_pb.Message) string {
 }
 
 func newTracerWrapper(tr pubsub.EventTracer, topics ...string) pubsub.EventTracer {
-	topicsMap := make(map[string]struct{})
-	for _, topic := range topics {
-		topicsMap[topic] = struct{}{}
+	var topicsMap map[string]struct{}
+	if len(topics) > 0 {
+		topicsMap = make(map[string]struct{})
+		for _, topic := range topics {
+			topicsMap[topic] = struct{}{}
+		}
 	}
+
 	return &tracerWrapper{tr: tr, topics: topicsMap}
 }
 
@@ -347,23 +343,42 @@ func (trw *tracerWrapper) Trace(evt *pubsub_pb.TraceEvent) {
 	// distributions.
 	// Furthermore, we only trace message publication and deliveries for specified topics
 	// (here just the blocks topic).
-	// TODO: hook all events into local metrics for inspection through the dashboard
 	switch evt.GetType() {
 	case pubsub_pb.TraceEvent_PUBLISH_MESSAGE:
-		if trw.traceMessage(evt.GetPublishMessage().Topics) {
+		stats.Record(context.TODO(), metrics.PubsubPublishMessage.M(1))
+		if trw.tr != nil && trw.traceMessage(evt.GetPublishMessage().Topics) {
 			trw.tr.Trace(evt)
 		}
 	case pubsub_pb.TraceEvent_DELIVER_MESSAGE:
-		if trw.traceMessage(evt.GetDeliverMessage().Topics) {
+		stats.Record(context.TODO(), metrics.PubsubDeliverMessage.M(1))
+		if trw.tr != nil && trw.traceMessage(evt.GetDeliverMessage().Topics) {
 			trw.tr.Trace(evt)
 		}
+	case pubsub_pb.TraceEvent_REJECT_MESSAGE:
+		stats.Record(context.TODO(), metrics.PubsubRejectMessage.M(1))
+	case pubsub_pb.TraceEvent_DUPLICATE_MESSAGE:
+		stats.Record(context.TODO(), metrics.PubsubDuplicateMessage.M(1))
 	case pubsub_pb.TraceEvent_JOIN:
-		trw.tr.Trace(evt)
+		if trw.tr != nil {
+			trw.tr.Trace(evt)
+		}
 	case pubsub_pb.TraceEvent_LEAVE:
-		trw.tr.Trace(evt)
+		if trw.tr != nil {
+			trw.tr.Trace(evt)
+		}
 	case pubsub_pb.TraceEvent_GRAFT:
-		trw.tr.Trace(evt)
+		if trw.tr != nil {
+			trw.tr.Trace(evt)
+		}
 	case pubsub_pb.TraceEvent_PRUNE:
-		trw.tr.Trace(evt)
+		if trw.tr != nil {
+			trw.tr.Trace(evt)
+		}
+	case pubsub_pb.TraceEvent_RECV_RPC:
+		stats.Record(context.TODO(), metrics.PubsubRecvRPC.M(1))
+	case pubsub_pb.TraceEvent_SEND_RPC:
+		stats.Record(context.TODO(), metrics.PubsubSendRPC.M(1))
+	case pubsub_pb.TraceEvent_DROP_RPC:
+		stats.Record(context.TODO(), metrics.PubsubDropRPC.M(1))
 	}
 }

node/modules/lp2p/transport.go

@@ -3,8 +3,8 @@ package lp2p
 import (
 	"github.com/libp2p/go-libp2p"
 	metrics "github.com/libp2p/go-libp2p-core/metrics"
+	noise "github.com/libp2p/go-libp2p-noise"
 	libp2pquic "github.com/libp2p/go-libp2p-quic-transport"
-	secio "github.com/libp2p/go-libp2p-secio"
 	tls "github.com/libp2p/go-libp2p-tls"
 )
 
@@ -23,9 +23,9 @@ func Security(enabled, preferTLS bool) interface{} {
 	}
 	return func() (opts Libp2pOpts) {
 		if preferTLS {
-			opts.Opts = append(opts.Opts, libp2p.ChainOptions(libp2p.Security(tls.ID, tls.New), libp2p.Security(secio.ID, secio.New)))
+			opts.Opts = append(opts.Opts, libp2p.ChainOptions(libp2p.Security(tls.ID, tls.New), libp2p.Security(noise.ID, noise.New)))
 		} else {
-			opts.Opts = append(opts.Opts, libp2p.ChainOptions(libp2p.Security(secio.ID, secio.New), libp2p.Security(tls.ID, tls.New)))
+			opts.Opts = append(opts.Opts, libp2p.ChainOptions(libp2p.Security(noise.ID, noise.New), libp2p.Security(tls.ID, tls.New)))
 		}
 		return opts
 	}

node/modules/services.go

@@ -80,13 +80,13 @@ func RunBlockSync(h host.Host, svc *blocksync.BlockSyncService) {
 func HandleIncomingBlocks(mctx helpers.MetricsCtx, lc fx.Lifecycle, ps *pubsub.PubSub, s *chain.Syncer, bserv dtypes.ChainBlockService, chain *store.ChainStore, stmgr *stmgr.StateManager, h host.Host, nn dtypes.NetworkName) {
 	ctx := helpers.LifecycleCtx(mctx, lc)
 
-	blocksub, err := ps.Subscribe(build.BlocksTopic(nn))
+	blocksub, err := ps.Subscribe(build.BlocksTopic(nn)) //nolint
 	if err != nil {
 		panic(err)
 	}
 
 	v := sub.NewBlockValidator(
-		chain, stmgr,
+		h.ID(), chain, stmgr,
 		func(p peer.ID) {
 			ps.BlacklistPeer(p)
 			h.ConnManager().TagPeer(p, "badblock", -1000)
@@ -99,15 +99,15 @@ func HandleIncomingBlocks(mctx helpers.MetricsCtx, lc fx.Lifecycle, ps *pubsub.P
 	go sub.HandleIncomingBlocks(ctx, blocksub, s, bserv, h.ConnManager())
 }
 
-func HandleIncomingMessages(mctx helpers.MetricsCtx, lc fx.Lifecycle, ps *pubsub.PubSub, mpool *messagepool.MessagePool, nn dtypes.NetworkName) {
+func HandleIncomingMessages(mctx helpers.MetricsCtx, lc fx.Lifecycle, ps *pubsub.PubSub, mpool *messagepool.MessagePool, h host.Host, nn dtypes.NetworkName) {
 	ctx := helpers.LifecycleCtx(mctx, lc)
 
-	msgsub, err := ps.Subscribe(build.MessagesTopic(nn))
+	msgsub, err := ps.Subscribe(build.MessagesTopic(nn)) //nolint:staticcheck
 	if err != nil {
 		panic(err)
 	}
 
-	v := sub.NewMessageValidator(mpool)
+	v := sub.NewMessageValidator(h.ID(), mpool)
 
 	if err := ps.RegisterTopicValidator(build.MessagesTopic(nn), v.Validate); err != nil {
 		panic(err)

node/modules/storageminer.go

@@ -36,20 +36,19 @@ import (
 	"github.com/filecoin-project/go-fil-markets/storagemarket"
 	storageimpl "github.com/filecoin-project/go-fil-markets/storagemarket/impl"
 	"github.com/filecoin-project/go-fil-markets/storagemarket/impl/funds"
-	"github.com/filecoin-project/go-fil-markets/storagemarket/impl/requestvalidation"
 	"github.com/filecoin-project/go-fil-markets/storagemarket/impl/storedask"
 	smnet "github.com/filecoin-project/go-fil-markets/storagemarket/network"
 	"github.com/filecoin-project/go-jsonrpc/auth"
 	"github.com/filecoin-project/go-multistore"
 	paramfetch "github.com/filecoin-project/go-paramfetch"
-	"github.com/filecoin-project/go-statestore"
 	"github.com/filecoin-project/go-storedcounter"
 	"github.com/filecoin-project/specs-actors/actors/abi"
 
-	sectorstorage "github.com/filecoin-project/sector-storage"
-	"github.com/filecoin-project/sector-storage/ffiwrapper"
-	"github.com/filecoin-project/sector-storage/stores"
-	sealing "github.com/filecoin-project/storage-fsm"
+	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"
+	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/build"
@@ -146,65 +145,68 @@ func SectorIDCounter(ds dtypes.MetadataDS) sealing.SectorIDCounter {
 type StorageMinerParams struct {
 	fx.In
 
-	Lifecycle         fx.Lifecycle
-	MetricsCtx        helpers.MetricsCtx
-	API               lapi.FullNode
-	Host              host.Host
-	MetadataDS        dtypes.MetadataDS
-	Sealer            sectorstorage.SectorManager
-	SectorIDCounter   sealing.SectorIDCounter
-	Verifier          ffiwrapper.Verifier
-	GetSealingDelayFn dtypes.GetSealingDelayFunc
+	Lifecycle          fx.Lifecycle
+	MetricsCtx         helpers.MetricsCtx
+	API                lapi.FullNode
+	Host               host.Host
+	MetadataDS         dtypes.MetadataDS
+	Sealer             sectorstorage.SectorManager
+	SectorIDCounter    sealing.SectorIDCounter
+	Verifier           ffiwrapper.Verifier
+	GetSealingConfigFn dtypes.GetSealingConfigFunc
 }
 
-func StorageMiner(params StorageMinerParams) (*storage.Miner, error) {
-	var (
-		ds     = params.MetadataDS
-		mctx   = params.MetricsCtx
-		lc     = params.Lifecycle
-		api    = params.API
-		sealer = params.Sealer
-		h      = params.Host
-		sc     = params.SectorIDCounter
-		verif  = params.Verifier
-		gsd    = params.GetSealingDelayFn
-	)
-	maddr, err := minerAddrFromDS(ds)
-	if err != nil {
-		return nil, err
+func StorageMiner(fc config.MinerFeeConfig) func(params StorageMinerParams) (*storage.Miner, error) {
+	return func(params StorageMinerParams) (*storage.Miner, error) {
+		var (
+			ds     = params.MetadataDS
+			mctx   = params.MetricsCtx
+			lc     = params.Lifecycle
+			api    = params.API
+			sealer = params.Sealer
+			h      = params.Host
+			sc     = params.SectorIDCounter
+			verif  = params.Verifier
+			gsd    = params.GetSealingConfigFn
+		)
+
+		maddr, err := minerAddrFromDS(ds)
+		if err != nil {
+			return nil, err
+		}
+
+		ctx := helpers.LifecycleCtx(mctx, lc)
+
+		mi, err := api.StateMinerInfo(ctx, maddr, types.EmptyTSK)
+		if err != nil {
+			return nil, err
+		}
+
+		worker, err := api.StateAccountKey(ctx, mi.Worker, types.EmptyTSK)
+		if err != nil {
+			return nil, err
+		}
+
+		fps, err := storage.NewWindowedPoStScheduler(api, fc, sealer, sealer, maddr, worker)
+		if err != nil {
+			return nil, err
+		}
+
+		sm, err := storage.NewMiner(api, maddr, worker, h, ds, sealer, sc, verif, gsd, fc)
+		if err != nil {
+			return nil, err
+		}
+
+		lc.Append(fx.Hook{
+			OnStart: func(context.Context) error {
+				go fps.Run(ctx)
+				return sm.Run(ctx)
+			},
+			OnStop: sm.Stop,
+		})
+
+		return sm, nil
 	}
-
-	ctx := helpers.LifecycleCtx(mctx, lc)
-
-	mi, err := api.StateMinerInfo(ctx, maddr, types.EmptyTSK)
-	if err != nil {
-		return nil, err
-	}
-
-	worker, err := api.StateAccountKey(ctx, mi.Worker, types.EmptyTSK)
-	if err != nil {
-		return nil, err
-	}
-
-	fps, err := storage.NewWindowedPoStScheduler(api, sealer, sealer, maddr, worker)
-	if err != nil {
-		return nil, err
-	}
-
-	sm, err := storage.NewMiner(api, maddr, worker, h, ds, sealer, sc, verif, gsd)
-	if err != nil {
-		return nil, err
-	}
-
-	lc.Append(fx.Hook{
-		OnStart: func(context.Context) error {
-			go fps.Run(ctx)
-			return sm.Run(ctx)
-		},
-		OnStop: sm.Stop,
-	})
-
-	return sm, nil
 }
 
 func HandleRetrieval(host host.Host, lc fx.Lifecycle, m retrievalmarket.RetrievalProvider) {
@@ -233,15 +235,6 @@ func HandleDeals(mctx helpers.MetricsCtx, lc fx.Lifecycle, host host.Host, h sto
 	})
 }
 
-// RegisterProviderValidator is an initialization hook that registers the provider
-// request validator with the data transfer module as the validator for
-// StorageDataTransferVoucher types
-func RegisterProviderValidator(mrv dtypes.ProviderRequestValidator, dtm dtypes.ProviderDataTransfer) {
-	if err := dtm.RegisterVoucherType(&requestvalidation.StorageDataTransferVoucher{}, (*requestvalidation.UnifiedRequestValidator)(mrv)); err != nil {
-		panic(err)
-	}
-}
-
 // NewProviderDAGServiceDataTransfer returns a data transfer manager that just
 // uses the provider's Staging DAG service for transfers
 func NewProviderDAGServiceDataTransfer(lc fx.Lifecycle, h host.Host, gs dtypes.StagingGraphsync, ds dtypes.MetadataDS) (dtypes.ProviderDataTransfer, error) {
@@ -266,11 +259,6 @@ func NewProviderDAGServiceDataTransfer(lc fx.Lifecycle, h host.Host, gs dtypes.S
 	return dt, nil
 }
 
-// NewProviderDealStore creates a statestore for the client to store its deals
-func NewProviderDealStore(ds dtypes.MetadataDS) dtypes.ProviderDealStore {
-	return statestore.New(namespace.Wrap(ds, datastore.NewKey("/deals/provider")))
-}
-
 // NewProviderPieceStore creates a statestore for storing metadata about pieces
 // shared by the storage and retrieval providers
 func NewProviderPieceStore(ds dtypes.MetadataDS) dtypes.ProviderPieceStore {
@@ -361,10 +349,6 @@ func SetupBlockProducer(lc fx.Lifecycle, ds dtypes.MetadataDS, api lapi.FullNode
 	return m, nil
 }
 
-func NewProviderRequestValidator(deals dtypes.ProviderDealStore) dtypes.ProviderRequestValidator {
-	return requestvalidation.NewUnifiedRequestValidator(deals, nil)
-}
-
 func NewStorageAsk(ctx helpers.MetricsCtx, fapi lapi.FullNode, ds dtypes.MetadataDS, minerAddress dtypes.MinerAddress, spn storagemarket.StorageProviderNode) (*storedask.StoredAsk, error) {
 
 	mi, err := fapi.StateMinerInfo(ctx, address.Address(minerAddress), types.EmptyTSK)
@@ -636,19 +620,28 @@ func NewSetConsiderOfflineRetrievalDealsConfigFunc(r repo.LockedRepo) (dtypes.Se
 	}, nil
 }
 
-func NewSetSealDelayFunc(r repo.LockedRepo) (dtypes.SetSealingDelayFunc, error) {
-	return func(delay time.Duration) (err error) {
-		err = mutateCfg(r, func(cfg *config.StorageMiner) {
-			cfg.SealingDelay = config.Duration(delay)
+func NewSetSealConfigFunc(r repo.LockedRepo) (dtypes.SetSealingConfigFunc, error) {
+	return func(cfg sealiface.Config) (err error) {
+		err = mutateCfg(r, func(c *config.StorageMiner) {
+			c.Sealing = config.SealingConfig{
+				MaxWaitDealsSectors: cfg.MaxWaitDealsSectors,
+				MaxSealingSectors:   cfg.MaxSealingSectors,
+				WaitDealsDelay:      config.Duration(cfg.WaitDealsDelay),
+			}
 		})
 		return
 	}, nil
 }
 
-func NewGetSealDelayFunc(r repo.LockedRepo) (dtypes.GetSealingDelayFunc, error) {
-	return func() (out time.Duration, err error) {
+func NewGetSealConfigFunc(r repo.LockedRepo) (dtypes.GetSealingConfigFunc, error) {
+	return func() (out sealiface.Config, err error) {
 		err = readCfg(r, func(cfg *config.StorageMiner) {
-			out = time.Duration(cfg.SealingDelay)
+			out = sealiface.Config{
+				MaxWaitDealsSectors:       cfg.Sealing.MaxWaitDealsSectors,
+				MaxSealingSectors:         cfg.Sealing.MaxSealingSectors,
+				MaxSealingSectorsForDeals: cfg.Sealing.MaxSealingSectorsForDeals,
+				WaitDealsDelay:            time.Duration(cfg.Sealing.WaitDealsDelay),
+			}
 		})
 		return
 	}, nil

node/node_test.go

@@ -1,55 +1,21 @@
 package node_test
 
 import (
-	"bytes"
-	"context"
-	"crypto/rand"
-	"io/ioutil"
-	"net/http/httptest"
 	"os"
-	"sync"
 	"testing"
 	"time"
 
+	builder "github.com/filecoin-project/lotus/node/test"
+
 	"github.com/filecoin-project/lotus/lib/lotuslog"
-	"github.com/filecoin-project/lotus/storage/mockstorage"
-	"github.com/filecoin-project/sector-storage/ffiwrapper"
-
-	"github.com/filecoin-project/go-storedcounter"
-	"github.com/ipfs/go-datastore"
-	logging "github.com/ipfs/go-log/v2"
-	"github.com/libp2p/go-libp2p-core/crypto"
-	"github.com/libp2p/go-libp2p-core/peer"
-	mocknet "github.com/libp2p/go-libp2p/p2p/net/mock"
-	"github.com/stretchr/testify/require"
-
-	"github.com/filecoin-project/go-address"
-	"github.com/filecoin-project/go-jsonrpc"
 	"github.com/filecoin-project/specs-actors/actors/abi"
 	"github.com/filecoin-project/specs-actors/actors/abi/big"
-	"github.com/filecoin-project/specs-actors/actors/builtin"
 	saminer "github.com/filecoin-project/specs-actors/actors/builtin/miner"
 	"github.com/filecoin-project/specs-actors/actors/builtin/power"
 	"github.com/filecoin-project/specs-actors/actors/builtin/verifreg"
+	logging "github.com/ipfs/go-log/v2"
 
-	"github.com/filecoin-project/lotus/api"
-	"github.com/filecoin-project/lotus/api/client"
 	"github.com/filecoin-project/lotus/api/test"
-	"github.com/filecoin-project/lotus/build"
-	"github.com/filecoin-project/lotus/chain/actors"
-	"github.com/filecoin-project/lotus/chain/gen"
-	genesis2 "github.com/filecoin-project/lotus/chain/gen/genesis"
-	"github.com/filecoin-project/lotus/chain/types"
-	"github.com/filecoin-project/lotus/chain/wallet"
-	"github.com/filecoin-project/lotus/cmd/lotus-seed/seed"
-	genesis "github.com/filecoin-project/lotus/genesis"
-	"github.com/filecoin-project/lotus/miner"
-	"github.com/filecoin-project/lotus/node"
-	"github.com/filecoin-project/lotus/node/modules"
-	modtest "github.com/filecoin-project/lotus/node/modules/testing"
-	"github.com/filecoin-project/lotus/node/repo"
-	sectorstorage "github.com/filecoin-project/sector-storage"
-	"github.com/filecoin-project/sector-storage/mock"
 )
 
 func init() {
@@ -62,435 +28,12 @@ func init() {
 	verifreg.MinVerifiedDealSize = big.NewInt(256)
 }
 
-func testStorageNode(ctx context.Context, t *testing.T, waddr address.Address, act address.Address, pk crypto.PrivKey, tnd test.TestNode, mn mocknet.Mocknet, opts node.Option) test.TestStorageNode {
-	r := repo.NewMemory(nil)
-
-	lr, err := r.Lock(repo.StorageMiner)
-	require.NoError(t, err)
-
-	ks, err := lr.KeyStore()
-	require.NoError(t, err)
-
-	kbytes, err := pk.Bytes()
-	require.NoError(t, err)
-
-	err = ks.Put("libp2p-host", types.KeyInfo{
-		Type:       "libp2p-host",
-		PrivateKey: kbytes,
-	})
-	require.NoError(t, err)
-
-	ds, err := lr.Datastore("/metadata")
-	require.NoError(t, err)
-	err = ds.Put(datastore.NewKey("miner-address"), act.Bytes())
-	require.NoError(t, err)
-
-	nic := storedcounter.New(ds, datastore.NewKey(modules.StorageCounterDSPrefix))
-	for i := 0; i < test.GenesisPreseals; i++ {
-		_, err := nic.Next()
-		require.NoError(t, err)
-	}
-	_, err = nic.Next()
-	require.NoError(t, err)
-
-	err = lr.Close()
-	require.NoError(t, err)
-
-	peerid, err := peer.IDFromPrivateKey(pk)
-	require.NoError(t, err)
-
-	enc, err := actors.SerializeParams(&saminer.ChangePeerIDParams{NewID: abi.PeerID(peerid)})
-	require.NoError(t, err)
-
-	msg := &types.Message{
-		To:     act,
-		From:   waddr,
-		Method: builtin.MethodsMiner.ChangePeerID,
-		Params: enc,
-		Value:  types.NewInt(0),
-	}
-
-	_, err = tnd.MpoolPushMessage(ctx, msg)
-	require.NoError(t, err)
-
-	// start node
-	var minerapi api.StorageMiner
-
-	mineBlock := make(chan miner.MineReq)
-	// TODO: use stop
-	_, err = node.New(ctx,
-		node.StorageMiner(&minerapi),
-		node.Online(),
-		node.Repo(r),
-		node.Test(),
-
-		node.MockHost(mn),
-
-		node.Override(new(api.FullNode), tnd),
-		node.Override(new(*miner.Miner), miner.NewTestMiner(mineBlock, act)),
-
-		opts,
-	)
-	if err != nil {
-		t.Fatalf("failed to construct node: %v", err)
-	}
-
-	/*// Bootstrap with full node
-	remoteAddrs, err := tnd.NetAddrsListen(ctx)
-	require.NoError(t, err)
-
-	err = minerapi.NetConnect(ctx, remoteAddrs)
-	require.NoError(t, err)*/
-	mineOne := func(ctx context.Context, req miner.MineReq) error {
-		select {
-		case mineBlock <- req:
-			return nil
-		case <-ctx.Done():
-			return ctx.Err()
-		}
-	}
-
-	return test.TestStorageNode{StorageMiner: minerapi, MineOne: mineOne}
-}
-
-func builder(t *testing.T, nFull int, storage []test.StorageMiner) ([]test.TestNode, []test.TestStorageNode) {
-	ctx := context.Background()
-	mn := mocknet.New(ctx)
-
-	fulls := make([]test.TestNode, nFull)
-	storers := make([]test.TestStorageNode, len(storage))
-
-	pk, _, err := crypto.GenerateEd25519Key(rand.Reader)
-	require.NoError(t, err)
-
-	minerPid, err := peer.IDFromPrivateKey(pk)
-	require.NoError(t, err)
-
-	var genbuf bytes.Buffer
-
-	if len(storage) > 1 {
-		panic("need more peer IDs")
-	}
-	// PRESEAL SECTION, TRY TO REPLACE WITH BETTER IN THE FUTURE
-	// TODO: would be great if there was a better way to fake the preseals
-
-	var genms []genesis.Miner
-	var maddrs []address.Address
-	var genaccs []genesis.Actor
-	var keys []*wallet.Key
-
-	var presealDirs []string
-	for i := 0; i < len(storage); i++ {
-		maddr, err := address.NewIDAddress(genesis2.MinerStart + uint64(i))
-		if err != nil {
-			t.Fatal(err)
-		}
-		tdir, err := ioutil.TempDir("", "preseal-memgen")
-		if err != nil {
-			t.Fatal(err)
-		}
-		genm, k, err := seed.PreSeal(maddr, abi.RegisteredSealProof_StackedDrg2KiBV1, 0, test.GenesisPreseals, tdir, []byte("make genesis mem random"), nil, true)
-		if err != nil {
-			t.Fatal(err)
-		}
-		genm.PeerId = minerPid
-
-		wk, err := wallet.NewKey(*k)
-		if err != nil {
-			return nil, nil
-		}
-
-		genaccs = append(genaccs, genesis.Actor{
-			Type:    genesis.TAccount,
-			Balance: big.Mul(big.NewInt(400_000_000), types.NewInt(build.FilecoinPrecision)),
-			Meta:    (&genesis.AccountMeta{Owner: wk.Address}).ActorMeta(),
-		})
-
-		keys = append(keys, wk)
-		presealDirs = append(presealDirs, tdir)
-		maddrs = append(maddrs, maddr)
-		genms = append(genms, *genm)
-	}
-	templ := &genesis.Template{
-		Accounts:        genaccs,
-		Miners:          genms,
-		Timestamp:       uint64(time.Now().Unix() - 10000), // some time sufficiently far in the past
-		VerifregRootKey: gen.DefaultVerifregRootkeyActor,
-	}
-
-	// END PRESEAL SECTION
-
-	for i := 0; i < nFull; i++ {
-		var genesis node.Option
-		if i == 0 {
-			genesis = node.Override(new(modules.Genesis), modtest.MakeGenesisMem(&genbuf, *templ))
-		} else {
-			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),
-			node.Online(),
-			node.Repo(repo.NewMemory(nil)),
-			node.MockHost(mn),
-			node.Test(),
-
-			genesis,
-		)
-		if err != nil {
-			t.Fatal(err)
-		}
-
-	}
-
-	for i, def := range storage {
-		// TODO: support non-bootstrap miners
-		if i != 0 {
-			t.Fatal("only one storage node supported")
-		}
-		if def.Full != 0 {
-			t.Fatal("storage nodes only supported on the first full node")
-		}
-
-		f := fulls[def.Full]
-		if _, err := f.FullNode.WalletImport(ctx, &keys[i].KeyInfo); err != nil {
-			t.Fatal(err)
-		}
-		if err := f.FullNode.WalletSetDefault(ctx, keys[i].Address); err != nil {
-			t.Fatal(err)
-		}
-
-		genMiner := maddrs[i]
-		wa := genms[i].Worker
-
-		storers[i] = testStorageNode(ctx, t, wa, genMiner, pk, f, mn, node.Options())
-		if err := storers[i].StorageAddLocal(ctx, presealDirs[i]); err != nil {
-			t.Fatalf("%+v", err)
-		}
-		/*
-			sma := storers[i].StorageMiner.(*impl.StorageMinerAPI)
-
-			psd := presealDirs[i]
-		*/
-	}
-
-	if err := mn.LinkAll(); err != nil {
-		t.Fatal(err)
-	}
-
-	if len(storers) > 0 {
-		// Mine 2 blocks to setup some CE stuff in some actors
-		var wait sync.Mutex
-		wait.Lock()
-
-		storers[0].MineOne(ctx, miner.MineReq{Done: func(bool, error) {
-			wait.Unlock()
-		}})
-		wait.Lock()
-		storers[0].MineOne(ctx, miner.MineReq{Done: func(bool, error) {
-			wait.Unlock()
-		}})
-		wait.Lock()
-	}
-
-	return fulls, storers
-}
-
-func mockSbBuilder(t *testing.T, nFull int, storage []test.StorageMiner) ([]test.TestNode, []test.TestStorageNode) {
-	ctx := context.Background()
-	mn := mocknet.New(ctx)
-
-	fulls := make([]test.TestNode, nFull)
-	storers := make([]test.TestStorageNode, len(storage))
-
-	var genbuf bytes.Buffer
-
-	// PRESEAL SECTION, TRY TO REPLACE WITH BETTER IN THE FUTURE
-	// TODO: would be great if there was a better way to fake the preseals
-
-	var genms []genesis.Miner
-	var genaccs []genesis.Actor
-	var maddrs []address.Address
-	var presealDirs []string
-	var keys []*wallet.Key
-	var pidKeys []crypto.PrivKey
-	for i := 0; i < len(storage); i++ {
-		maddr, err := address.NewIDAddress(genesis2.MinerStart + uint64(i))
-		if err != nil {
-			t.Fatal(err)
-		}
-		tdir, err := ioutil.TempDir("", "preseal-memgen")
-		if err != nil {
-			t.Fatal(err)
-		}
-
-		preseals := storage[i].Preseal
-		if preseals == test.PresealGenesis {
-			preseals = test.GenesisPreseals
-		}
-
-		genm, k, err := mockstorage.PreSeal(2048, maddr, preseals)
-		if err != nil {
-			t.Fatal(err)
-		}
-
-		pk, _, err := crypto.GenerateEd25519Key(rand.Reader)
-		require.NoError(t, err)
-
-		minerPid, err := peer.IDFromPrivateKey(pk)
-		require.NoError(t, err)
-
-		genm.PeerId = minerPid
-
-		wk, err := wallet.NewKey(*k)
-		if err != nil {
-			return nil, nil
-		}
-
-		genaccs = append(genaccs, genesis.Actor{
-			Type:    genesis.TAccount,
-			Balance: big.Mul(big.NewInt(400_000_000_000), types.NewInt(build.FilecoinPrecision)),
-			Meta:    (&genesis.AccountMeta{Owner: wk.Address}).ActorMeta(),
-		})
-
-		keys = append(keys, wk)
-		pidKeys = append(pidKeys, pk)
-		presealDirs = append(presealDirs, tdir)
-		maddrs = append(maddrs, maddr)
-		genms = append(genms, *genm)
-	}
-	templ := &genesis.Template{
-		Accounts:        genaccs,
-		Miners:          genms,
-		Timestamp:       uint64(time.Now().Unix()) - (build.BlockDelaySecs * 20000),
-		VerifregRootKey: gen.DefaultVerifregRootkeyActor,
-	}
-
-	// END PRESEAL SECTION
-
-	for i := 0; i < nFull; i++ {
-		var genesis node.Option
-		if i == 0 {
-			genesis = node.Override(new(modules.Genesis), modtest.MakeGenesisMem(&genbuf, *templ))
-		} else {
-			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),
-			node.Online(),
-			node.Repo(repo.NewMemory(nil)),
-			node.MockHost(mn),
-			node.Test(),
-
-			node.Override(new(ffiwrapper.Verifier), mock.MockVerifier),
-
-			genesis,
-		)
-		if err != nil {
-			t.Fatalf("%+v", err)
-		}
-	}
-
-	for i, def := range storage {
-		// TODO: support non-bootstrap miners
-
-		minerID := abi.ActorID(genesis2.MinerStart + uint64(i))
-
-		if def.Full != 0 {
-			t.Fatal("storage nodes only supported on the first full node")
-		}
-
-		f := fulls[def.Full]
-		if _, err := f.FullNode.WalletImport(ctx, &keys[i].KeyInfo); err != nil {
-			return nil, nil
-		}
-		if err := f.FullNode.WalletSetDefault(ctx, keys[i].Address); err != nil {
-			return nil, nil
-		}
-
-		sectors := make([]abi.SectorID, len(genms[i].Sectors))
-		for i, sector := range genms[i].Sectors {
-			sectors[i] = abi.SectorID{
-				Miner:  minerID,
-				Number: sector.SectorID,
-			}
-		}
-
-		storers[i] = testStorageNode(ctx, t, genms[i].Worker, maddrs[i], pidKeys[i], f, mn, node.Options(
-			node.Override(new(sectorstorage.SectorManager), func() (sectorstorage.SectorManager, error) {
-				return mock.NewMockSectorMgr(build.DefaultSectorSize(), sectors), nil
-			}),
-			node.Override(new(ffiwrapper.Verifier), mock.MockVerifier),
-			node.Unset(new(*sectorstorage.Manager)),
-		))
-	}
-
-	if err := mn.LinkAll(); err != nil {
-		t.Fatal(err)
-	}
-
-	if len(storers) > 0 {
-		// Mine 2 blocks to setup some CE stuff in some actors
-		var wait sync.Mutex
-		wait.Lock()
-
-		storers[0].MineOne(ctx, miner.MineReq{Done: func(bool, error) {
-			wait.Unlock()
-		}})
-		wait.Lock()
-		storers[0].MineOne(ctx, miner.MineReq{Done: func(bool, error) {
-			wait.Unlock()
-		}})
-		wait.Lock()
-	}
-
-	return fulls, storers
-}
-
 func TestAPI(t *testing.T) {
-	test.TestApis(t, builder)
-}
-
-func rpcBuilder(t *testing.T, nFull int, storage []test.StorageMiner) ([]test.TestNode, []test.TestStorageNode) {
-	fullApis, storaApis := builder(t, nFull, storage)
-	fulls := make([]test.TestNode, nFull)
-	storers := make([]test.TestStorageNode, len(storage))
-
-	for i, a := range fullApis {
-		rpcServer := jsonrpc.NewServer()
-		rpcServer.Register("Filecoin", a)
-		testServ := httptest.NewServer(rpcServer) //  todo: close
-
-		var err error
-		fulls[i].FullNode, _, err = client.NewFullNodeRPC("ws://"+testServ.Listener.Addr().String(), nil)
-		if err != nil {
-			t.Fatal(err)
-		}
-	}
-
-	for i, a := range storaApis {
-		rpcServer := jsonrpc.NewServer()
-		rpcServer.Register("Filecoin", a)
-		testServ := httptest.NewServer(rpcServer) //  todo: close
-
-		var err error
-		storers[i].StorageMiner, _, err = client.NewStorageMinerRPC("ws://"+testServ.Listener.Addr().String(), nil)
-		if err != nil {
-			t.Fatal(err)
-		}
-		storers[i].MineOne = a.MineOne
-	}
-
-	return fulls, storers
+	test.TestApis(t, builder.Builder)
 }
 
 func TestAPIRPC(t *testing.T) {
-	test.TestApis(t, rpcBuilder)
+	test.TestApis(t, builder.RPCBuilder)
 }
 
 func TestAPIDealFlow(t *testing.T) {
@@ -501,16 +44,16 @@ func TestAPIDealFlow(t *testing.T) {
 	logging.SetLogLevel("storageminer", "ERROR")
 
 	t.Run("TestDealFlow", func(t *testing.T) {
-		test.TestDealFlow(t, mockSbBuilder, 10*time.Millisecond, false, false)
+		test.TestDealFlow(t, builder.MockSbBuilder, 10*time.Millisecond, false, false)
 	})
 	t.Run("WithExportedCAR", func(t *testing.T) {
-		test.TestDealFlow(t, mockSbBuilder, 10*time.Millisecond, true, false)
+		test.TestDealFlow(t, builder.MockSbBuilder, 10*time.Millisecond, true, false)
 	})
 	t.Run("TestDoubleDealFlow", func(t *testing.T) {
-		test.TestDoubleDealFlow(t, mockSbBuilder, 10*time.Millisecond)
+		test.TestDoubleDealFlow(t, builder.MockSbBuilder, 10*time.Millisecond)
 	})
 	t.Run("TestFastRetrievalDealFlow", func(t *testing.T) {
-		test.TestFastRetrievalDealFlow(t, mockSbBuilder, 10*time.Millisecond)
+		test.TestFastRetrievalDealFlow(t, builder.MockSbBuilder, 10*time.Millisecond)
 	})
 }
 
@@ -528,15 +71,15 @@ func TestAPIDealFlowReal(t *testing.T) {
 	saminer.PreCommitChallengeDelay = 5
 
 	t.Run("basic", func(t *testing.T) {
-		test.TestDealFlow(t, builder, time.Second, false, false)
+		test.TestDealFlow(t, builder.Builder, time.Second, false, false)
 	})
 
 	t.Run("fast-retrieval", func(t *testing.T) {
-		test.TestDealFlow(t, builder, time.Second, false, true)
+		test.TestDealFlow(t, builder.Builder, time.Second, false, true)
 	})
 
 	t.Run("retrieval-second", func(t *testing.T) {
-		test.TestSenondDealRetrieval(t, builder, time.Second)
+		test.TestSenondDealRetrieval(t, builder.Builder, time.Second)
 	})
 }
 
@@ -547,7 +90,7 @@ func TestDealMining(t *testing.T) {
 	logging.SetLogLevel("sub", "ERROR")
 	logging.SetLogLevel("storageminer", "ERROR")
 
-	test.TestDealMining(t, mockSbBuilder, 50*time.Millisecond, false)
+	test.TestDealMining(t, builder.MockSbBuilder, 50*time.Millisecond, false)
 }
 
 func TestPledgeSectors(t *testing.T) {
@@ -558,11 +101,11 @@ func TestPledgeSectors(t *testing.T) {
 	logging.SetLogLevel("storageminer", "ERROR")
 
 	t.Run("1", func(t *testing.T) {
-		test.TestPledgeSector(t, mockSbBuilder, 50*time.Millisecond, 1)
+		test.TestPledgeSector(t, builder.MockSbBuilder, 50*time.Millisecond, 1)
 	})
 
 	t.Run("100", func(t *testing.T) {
-		test.TestPledgeSector(t, mockSbBuilder, 50*time.Millisecond, 100)
+		test.TestPledgeSector(t, builder.MockSbBuilder, 50*time.Millisecond, 100)
 	})
 
 	t.Run("1000", func(t *testing.T) {
@@ -570,7 +113,7 @@ func TestPledgeSectors(t *testing.T) {
 			t.Skip("skipping test in short mode")
 		}
 
-		test.TestPledgeSector(t, mockSbBuilder, 50*time.Millisecond, 1000)
+		test.TestPledgeSector(t, builder.MockSbBuilder, 50*time.Millisecond, 1000)
 	})
 }
 
@@ -585,7 +128,7 @@ func TestWindowedPost(t *testing.T) {
 	logging.SetLogLevel("sub", "ERROR")
 	logging.SetLogLevel("storageminer", "ERROR")
 
-	test.TestWindowPost(t, mockSbBuilder, 2*time.Millisecond, 10)
+	test.TestWindowPost(t, builder.MockSbBuilder, 2*time.Millisecond, 10)
 }
 
 func TestCCUpgrade(t *testing.T) {
@@ -595,7 +138,7 @@ func TestCCUpgrade(t *testing.T) {
 	logging.SetLogLevel("sub", "ERROR")
 	logging.SetLogLevel("storageminer", "ERROR")
 
-	test.TestCCUpgrade(t, mockSbBuilder, 5*time.Millisecond)
+	test.TestCCUpgrade(t, builder.MockSbBuilder, 5*time.Millisecond)
 }
 
 func TestPaymentChannels(t *testing.T) {
@@ -606,5 +149,5 @@ func TestPaymentChannels(t *testing.T) {
 	logging.SetLogLevel("pubsub", "ERROR")
 	logging.SetLogLevel("storageminer", "ERROR")
 
-	test.TestPaymentChannels(t, mockSbBuilder, 5*time.Millisecond)
+	test.TestPaymentChannels(t, builder.MockSbBuilder, 5*time.Millisecond)
 }

node/repo/fsrepo.go

@@ -20,8 +20,8 @@ import (
 	"github.com/multiformats/go-multiaddr"
 	"golang.org/x/xerrors"
 
-	"github.com/filecoin-project/sector-storage/fsutil"
-	"github.com/filecoin-project/sector-storage/stores"
+	"github.com/filecoin-project/lotus/extern/sector-storage/fsutil"
+	"github.com/filecoin-project/lotus/extern/sector-storage/stores"
 
 	"github.com/filecoin-project/lotus/chain/types"
 	"github.com/filecoin-project/lotus/node/config"

node/repo/interface.go

@@ -6,8 +6,8 @@ import (
 	"github.com/ipfs/go-datastore"
 	"github.com/multiformats/go-multiaddr"
 
-	"github.com/filecoin-project/sector-storage/fsutil"
-	"github.com/filecoin-project/sector-storage/stores"
+	"github.com/filecoin-project/lotus/extern/sector-storage/fsutil"
+	"github.com/filecoin-project/lotus/extern/sector-storage/stores"
 
 	"github.com/filecoin-project/lotus/chain/types"
 )

node/repo/memrepo.go

@@ -14,11 +14,11 @@ import (
 	"github.com/multiformats/go-multiaddr"
 	"golang.org/x/xerrors"
 
-	"github.com/filecoin-project/sector-storage/fsutil"
+	"github.com/filecoin-project/lotus/extern/sector-storage/fsutil"
 
 	"github.com/filecoin-project/lotus/chain/types"
+	"github.com/filecoin-project/lotus/extern/sector-storage/stores"
 	"github.com/filecoin-project/lotus/node/config"
-	"github.com/filecoin-project/sector-storage/stores"
 )
 
 type MemRepo struct {

node/test/builder.go

@@ -0,0 +1,500 @@
+package test
+
+import (
+	"bytes"
+	"context"
+	"crypto/rand"
+	"io/ioutil"
+	"net"
+	"net/http/httptest"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/filecoin-project/go-address"
+	"github.com/filecoin-project/go-jsonrpc"
+	"github.com/filecoin-project/go-storedcounter"
+	"github.com/filecoin-project/lotus/api"
+	"github.com/filecoin-project/lotus/api/client"
+	"github.com/filecoin-project/lotus/api/test"
+	"github.com/filecoin-project/lotus/build"
+	"github.com/filecoin-project/lotus/chain/actors"
+	"github.com/filecoin-project/lotus/chain/gen"
+	genesis2 "github.com/filecoin-project/lotus/chain/gen/genesis"
+	"github.com/filecoin-project/lotus/chain/types"
+	"github.com/filecoin-project/lotus/chain/wallet"
+	"github.com/filecoin-project/lotus/cmd/lotus-seed/seed"
+	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/mock"
+	"github.com/filecoin-project/lotus/genesis"
+	miner2 "github.com/filecoin-project/lotus/miner"
+	"github.com/filecoin-project/lotus/node"
+	"github.com/filecoin-project/lotus/node/modules"
+	testing2 "github.com/filecoin-project/lotus/node/modules/testing"
+	"github.com/filecoin-project/lotus/node/repo"
+	"github.com/filecoin-project/lotus/storage/mockstorage"
+	"github.com/filecoin-project/specs-actors/actors/abi"
+	"github.com/filecoin-project/specs-actors/actors/abi/big"
+	"github.com/filecoin-project/specs-actors/actors/builtin"
+	"github.com/filecoin-project/specs-actors/actors/builtin/miner"
+	"github.com/ipfs/go-datastore"
+	"github.com/libp2p/go-libp2p-core/crypto"
+	"github.com/libp2p/go-libp2p-core/peer"
+	mocknet "github.com/libp2p/go-libp2p/p2p/net/mock"
+	"github.com/multiformats/go-multiaddr"
+	"github.com/stretchr/testify/require"
+)
+
+func CreateTestStorageNode(ctx context.Context, t *testing.T, waddr address.Address, act address.Address, pk crypto.PrivKey, tnd test.TestNode, mn mocknet.Mocknet, opts node.Option) test.TestStorageNode {
+	r := repo.NewMemory(nil)
+
+	lr, err := r.Lock(repo.StorageMiner)
+	require.NoError(t, err)
+
+	ks, err := lr.KeyStore()
+	require.NoError(t, err)
+
+	kbytes, err := pk.Bytes()
+	require.NoError(t, err)
+
+	err = ks.Put("libp2p-host", types.KeyInfo{
+		Type:       "libp2p-host",
+		PrivateKey: kbytes,
+	})
+	require.NoError(t, err)
+
+	ds, err := lr.Datastore("/metadata")
+	require.NoError(t, err)
+	err = ds.Put(datastore.NewKey("miner-address"), act.Bytes())
+	require.NoError(t, err)
+
+	nic := storedcounter.New(ds, datastore.NewKey(modules.StorageCounterDSPrefix))
+	for i := 0; i < test.GenesisPreseals; i++ {
+		_, err := nic.Next()
+		require.NoError(t, err)
+	}
+	_, err = nic.Next()
+	require.NoError(t, err)
+
+	err = lr.Close()
+	require.NoError(t, err)
+
+	peerid, err := peer.IDFromPrivateKey(pk)
+	require.NoError(t, err)
+
+	enc, err := actors.SerializeParams(&miner.ChangePeerIDParams{NewID: abi.PeerID(peerid)})
+	require.NoError(t, err)
+
+	msg := &types.Message{
+		To:     act,
+		From:   waddr,
+		Method: builtin.MethodsMiner.ChangePeerID,
+		Params: enc,
+		Value:  types.NewInt(0),
+	}
+
+	_, err = tnd.MpoolPushMessage(ctx, msg, nil)
+	require.NoError(t, err)
+
+	// start node
+	var minerapi api.StorageMiner
+
+	mineBlock := make(chan miner2.MineReq)
+	// TODO: use stop
+	_, err = node.New(ctx,
+		node.StorageMiner(&minerapi),
+		node.Online(),
+		node.Repo(r),
+		node.Test(),
+
+		node.MockHost(mn),
+
+		node.Override(new(api.FullNode), tnd),
+		node.Override(new(*miner2.Miner), miner2.NewTestMiner(mineBlock, act)),
+
+		opts,
+	)
+	if err != nil {
+		t.Fatalf("failed to construct node: %v", err)
+	}
+
+	/*// Bootstrap with full node
+	remoteAddrs, err := tnd.NetAddrsListen(ctx)
+	require.NoError(t, err)
+
+	err = minerapi.NetConnect(ctx, remoteAddrs)
+	require.NoError(t, err)*/
+	mineOne := func(ctx context.Context, req miner2.MineReq) error {
+		select {
+		case mineBlock <- req:
+			return nil
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}
+
+	return test.TestStorageNode{StorageMiner: minerapi, MineOne: mineOne}
+}
+
+func Builder(t *testing.T, nFull int, storage []test.StorageMiner) ([]test.TestNode, []test.TestStorageNode) {
+	ctx := context.Background()
+	mn := mocknet.New(ctx)
+
+	fulls := make([]test.TestNode, nFull)
+	storers := make([]test.TestStorageNode, len(storage))
+
+	pk, _, err := crypto.GenerateEd25519Key(rand.Reader)
+	require.NoError(t, err)
+
+	minerPid, err := peer.IDFromPrivateKey(pk)
+	require.NoError(t, err)
+
+	var genbuf bytes.Buffer
+
+	if len(storage) > 1 {
+		panic("need more peer IDs")
+	}
+	// PRESEAL SECTION, TRY TO REPLACE WITH BETTER IN THE FUTURE
+	// TODO: would be great if there was a better way to fake the preseals
+
+	var genms []genesis.Miner
+	var maddrs []address.Address
+	var genaccs []genesis.Actor
+	var keys []*wallet.Key
+
+	var presealDirs []string
+	for i := 0; i < len(storage); i++ {
+		maddr, err := address.NewIDAddress(genesis2.MinerStart + uint64(i))
+		if err != nil {
+			t.Fatal(err)
+		}
+		tdir, err := ioutil.TempDir("", "preseal-memgen")
+		if err != nil {
+			t.Fatal(err)
+		}
+		genm, k, err := seed.PreSeal(maddr, abi.RegisteredSealProof_StackedDrg2KiBV1, 0, test.GenesisPreseals, tdir, []byte("make genesis mem random"), nil, true)
+		if err != nil {
+			t.Fatal(err)
+		}
+		genm.PeerId = minerPid
+
+		wk, err := wallet.NewKey(*k)
+		if err != nil {
+			return nil, nil
+		}
+
+		genaccs = append(genaccs, genesis.Actor{
+			Type:    genesis.TAccount,
+			Balance: big.Mul(big.NewInt(400000000), types.NewInt(build.FilecoinPrecision)),
+			Meta:    (&genesis.AccountMeta{Owner: wk.Address}).ActorMeta(),
+		})
+
+		keys = append(keys, wk)
+		presealDirs = append(presealDirs, tdir)
+		maddrs = append(maddrs, maddr)
+		genms = append(genms, *genm)
+	}
+	templ := &genesis.Template{
+		Accounts:         genaccs,
+		Miners:           genms,
+		Timestamp:        uint64(time.Now().Unix() - 10000), // some time sufficiently far in the past
+		VerifregRootKey:  gen.DefaultVerifregRootkeyActor,
+		RemainderAccount: gen.DefaultRemainderAccountActor,
+	}
+
+	// END PRESEAL SECTION
+
+	for i := 0; i < nFull; i++ {
+		var genesis node.Option
+		if i == 0 {
+			genesis = node.Override(new(modules.Genesis), testing2.MakeGenesisMem(&genbuf, *templ))
+		} else {
+			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),
+			node.Online(),
+			node.Repo(repo.NewMemory(nil)),
+			node.MockHost(mn),
+			node.Test(),
+
+			genesis,
+		)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+	}
+
+	for i, def := range storage {
+		// TODO: support non-bootstrap miners
+		if i != 0 {
+			t.Fatal("only one storage node supported")
+		}
+		if def.Full != 0 {
+			t.Fatal("storage nodes only supported on the first full node")
+		}
+
+		f := fulls[def.Full]
+		if _, err := f.FullNode.WalletImport(ctx, &keys[i].KeyInfo); err != nil {
+			t.Fatal(err)
+		}
+		if err := f.FullNode.WalletSetDefault(ctx, keys[i].Address); err != nil {
+			t.Fatal(err)
+		}
+
+		genMiner := maddrs[i]
+		wa := genms[i].Worker
+
+		storers[i] = CreateTestStorageNode(ctx, t, wa, genMiner, pk, f, mn, node.Options())
+		if err := storers[i].StorageAddLocal(ctx, presealDirs[i]); err != nil {
+			t.Fatalf("%+v", err)
+		}
+		/*
+			sma := storers[i].StorageMiner.(*impl.StorageMinerAPI)
+
+			psd := presealDirs[i]
+		*/
+	}
+
+	if err := mn.LinkAll(); err != nil {
+		t.Fatal(err)
+	}
+
+	if len(storers) > 0 {
+		// Mine 2 blocks to setup some CE stuff in some actors
+		var wait sync.Mutex
+		wait.Lock()
+
+		test.MineUntilBlock(ctx, t, storers[0], func() {
+			wait.Unlock()
+		})
+		wait.Lock()
+		test.MineUntilBlock(ctx, t, storers[0], func() {
+			wait.Unlock()
+		})
+		wait.Lock()
+	}
+
+	return fulls, storers
+}
+
+func MockSbBuilder(t *testing.T, nFull int, storage []test.StorageMiner) ([]test.TestNode, []test.TestStorageNode) {
+	ctx := context.Background()
+	mn := mocknet.New(ctx)
+
+	fulls := make([]test.TestNode, nFull)
+	storers := make([]test.TestStorageNode, len(storage))
+
+	var genbuf bytes.Buffer
+
+	// PRESEAL SECTION, TRY TO REPLACE WITH BETTER IN THE FUTURE
+	// TODO: would be great if there was a better way to fake the preseals
+
+	var genms []genesis.Miner
+	var genaccs []genesis.Actor
+	var maddrs []address.Address
+	var keys []*wallet.Key
+	var pidKeys []crypto.PrivKey
+	for i := 0; i < len(storage); i++ {
+		maddr, err := address.NewIDAddress(genesis2.MinerStart + uint64(i))
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		preseals := storage[i].Preseal
+		if preseals == test.PresealGenesis {
+			preseals = test.GenesisPreseals
+		}
+
+		genm, k, err := mockstorage.PreSeal(2048, maddr, preseals)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		pk, _, err := crypto.GenerateEd25519Key(rand.Reader)
+		require.NoError(t, err)
+
+		minerPid, err := peer.IDFromPrivateKey(pk)
+		require.NoError(t, err)
+
+		genm.PeerId = minerPid
+
+		wk, err := wallet.NewKey(*k)
+		if err != nil {
+			return nil, nil
+		}
+
+		genaccs = append(genaccs, genesis.Actor{
+			Type:    genesis.TAccount,
+			Balance: big.Mul(big.NewInt(400000000), types.NewInt(build.FilecoinPrecision)),
+			Meta:    (&genesis.AccountMeta{Owner: wk.Address}).ActorMeta(),
+		})
+
+		keys = append(keys, wk)
+		pidKeys = append(pidKeys, pk)
+		maddrs = append(maddrs, maddr)
+		genms = append(genms, *genm)
+	}
+	templ := &genesis.Template{
+		Accounts:         genaccs,
+		Miners:           genms,
+		Timestamp:        uint64(time.Now().Unix()) - (build.BlockDelaySecs * 20000),
+		VerifregRootKey:  gen.DefaultVerifregRootkeyActor,
+		RemainderAccount: gen.DefaultRemainderAccountActor,
+	}
+
+	// END PRESEAL SECTION
+
+	for i := 0; i < nFull; i++ {
+		var genesis node.Option
+		if i == 0 {
+			genesis = node.Override(new(modules.Genesis), testing2.MakeGenesisMem(&genbuf, *templ))
+		} else {
+			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),
+			node.Online(),
+			node.Repo(repo.NewMemory(nil)),
+			node.MockHost(mn),
+			node.Test(),
+
+			node.Override(new(ffiwrapper.Verifier), mock.MockVerifier),
+
+			genesis,
+		)
+		if err != nil {
+			t.Fatalf("%+v", err)
+		}
+	}
+
+	for i, def := range storage {
+		// TODO: support non-bootstrap miners
+
+		minerID := abi.ActorID(genesis2.MinerStart + uint64(i))
+
+		if def.Full != 0 {
+			t.Fatal("storage nodes only supported on the first full node")
+		}
+
+		f := fulls[def.Full]
+		if _, err := f.FullNode.WalletImport(ctx, &keys[i].KeyInfo); err != nil {
+			return nil, nil
+		}
+		if err := f.FullNode.WalletSetDefault(ctx, keys[i].Address); err != nil {
+			return nil, nil
+		}
+
+		sectors := make([]abi.SectorID, len(genms[i].Sectors))
+		for i, sector := range genms[i].Sectors {
+			sectors[i] = abi.SectorID{
+				Miner:  minerID,
+				Number: sector.SectorID,
+			}
+		}
+
+		storers[i] = CreateTestStorageNode(ctx, t, genms[i].Worker, maddrs[i], pidKeys[i], f, mn, node.Options(
+			node.Override(new(sectorstorage.SectorManager), func() (sectorstorage.SectorManager, error) {
+				return mock.NewMockSectorMgr(build.DefaultSectorSize(), sectors), nil
+			}),
+			node.Override(new(ffiwrapper.Verifier), mock.MockVerifier),
+			node.Unset(new(*sectorstorage.Manager)),
+		))
+	}
+
+	if err := mn.LinkAll(); err != nil {
+		t.Fatal(err)
+	}
+
+	if len(storers) > 0 {
+		// Mine 2 blocks to setup some CE stuff in some actors
+		var wait sync.Mutex
+		wait.Lock()
+
+		test.MineUntilBlock(ctx, t, storers[0], func() {
+			wait.Unlock()
+		})
+		wait.Lock()
+		test.MineUntilBlock(ctx, t, storers[0], func() {
+			wait.Unlock()
+		})
+		wait.Lock()
+	}
+
+	return fulls, storers
+}
+
+func RPCBuilder(t *testing.T, nFull int, storage []test.StorageMiner) ([]test.TestNode, []test.TestStorageNode) {
+	return rpcWithBuilder(t, Builder, nFull, storage)
+}
+
+func RPCMockSbBuilder(t *testing.T, nFull int, storage []test.StorageMiner) ([]test.TestNode, []test.TestStorageNode) {
+	return rpcWithBuilder(t, MockSbBuilder, nFull, storage)
+}
+
+func rpcWithBuilder(t *testing.T, b test.APIBuilder, nFull int, storage []test.StorageMiner) ([]test.TestNode, []test.TestStorageNode) {
+	fullApis, storaApis := b(t, nFull, storage)
+	fulls := make([]test.TestNode, nFull)
+	storers := make([]test.TestStorageNode, len(storage))
+
+	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
+	}
+
+	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
+}
+
+func parseWSSMultiAddr(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")
+	if err != nil {
+		return nil, err
+	}
+	return ma, nil
+}