Merge branch 'master' into libp2p-pubsub-tracer

node/modules/alerts.go

@@ -0,0 +1,44 @@
+package modules
+
+import (
+	"github.com/filecoin-project/lotus/journal/alerting"
+	"github.com/filecoin-project/lotus/lib/ulimit"
+)
+
+func CheckFdLimit(min uint64) func(al *alerting.Alerting) {
+	return func(al *alerting.Alerting) {
+		soft, _, err := ulimit.GetLimit()
+
+		if err == ulimit.ErrUnsupported {
+			log.Warn("FD limit monitoring not available")
+			return
+		}
+
+		alert := al.AddAlertType("process", "fd-limit")
+		if err != nil {
+			al.Raise(alert, map[string]string{
+				"message": "failed to get FD limit",
+				"error":   err.Error(),
+			})
+		}
+
+		if soft < min {
+			al.Raise(alert, map[string]interface{}{
+				"message":         "soft FD limit is low",
+				"soft_limit":      soft,
+				"recommended_min": min,
+			})
+		}
+	}
+}
+
+// TODO: More things:
+//  * Space in repo dirs (taking into account mounts)
+//  * Miner
+//    * Faulted partitions
+//    * Low balances
+//  * Market provider
+//    * Reachability
+//    * on-chain config
+//  * Low memory (maybe)
+//  * Network / sync issues

node/modules/chain.go

@@ -17,13 +17,13 @@ import (
 	"github.com/filecoin-project/lotus/build"
 	"github.com/filecoin-project/lotus/chain"
 	"github.com/filecoin-project/lotus/chain/beacon"
+	"github.com/filecoin-project/lotus/chain/consensus"
 	"github.com/filecoin-project/lotus/chain/exchange"
 	"github.com/filecoin-project/lotus/chain/gen/slashfilter"
 	"github.com/filecoin-project/lotus/chain/messagepool"
 	"github.com/filecoin-project/lotus/chain/stmgr"
 	"github.com/filecoin-project/lotus/chain/store"
 	"github.com/filecoin-project/lotus/chain/vm"
-	"github.com/filecoin-project/lotus/extern/sector-storage/ffiwrapper"
 	"github.com/filecoin-project/lotus/journal"
 	"github.com/filecoin-project/lotus/node/modules/dtypes"
 	"github.com/filecoin-project/lotus/node/modules/helpers"
@@ -58,8 +58,8 @@ func ChainBlockService(bs dtypes.ExposedBlockstore, rem dtypes.ChainBitswap) dty
 	return blockservice.New(bs, rem)
 }
 
-func MessagePool(lc fx.Lifecycle, mpp messagepool.Provider, ds dtypes.MetadataDS, nn dtypes.NetworkName, j journal.Journal, protector dtypes.GCReferenceProtector) (*messagepool.MessagePool, error) {
-	mp, err := messagepool.New(mpp, ds, nn, j)
+func MessagePool(lc fx.Lifecycle, us stmgr.UpgradeSchedule, mpp messagepool.Provider, ds dtypes.MetadataDS, nn dtypes.NetworkName, j journal.Journal, protector dtypes.GCReferenceProtector) (*messagepool.MessagePool, error) {
+	mp, err := messagepool.New(mpp, ds, us, nn, j)
 	if err != nil {
 		return nil, xerrors.Errorf("constructing mpool: %w", err)
 	}
@@ -72,8 +72,15 @@ func MessagePool(lc fx.Lifecycle, mpp messagepool.Provider, ds dtypes.MetadataDS
 	return mp, nil
 }
 
-func ChainStore(lc fx.Lifecycle, cbs dtypes.ChainBlockstore, sbs dtypes.StateBlockstore, ds dtypes.MetadataDS, basebs dtypes.BaseBlockstore, j journal.Journal) *store.ChainStore {
-	chain := store.NewChainStore(cbs, sbs, ds, j)
+func ChainStore(lc fx.Lifecycle,
+	cbs dtypes.ChainBlockstore,
+	sbs dtypes.StateBlockstore,
+	ds dtypes.MetadataDS,
+	basebs dtypes.BaseBlockstore,
+	weight store.WeightFunc,
+	j journal.Journal) *store.ChainStore {
+
+	chain := store.NewChainStore(cbs, sbs, ds, weight, j)
 
 	if err := chain.Load(); err != nil {
 		log.Warnf("loading chain state from disk: %s", err)
@@ -100,14 +107,20 @@ func ChainStore(lc fx.Lifecycle, cbs dtypes.ChainBlockstore, sbs dtypes.StateBlo
 	return chain
 }
 
-func NetworkName(mctx helpers.MetricsCtx, lc fx.Lifecycle, cs *store.ChainStore, syscalls vm.SyscallBuilder, us stmgr.UpgradeSchedule, _ dtypes.AfterGenesisSet) (dtypes.NetworkName, error) {
+func NetworkName(mctx helpers.MetricsCtx,
+	lc fx.Lifecycle,
+	cs *store.ChainStore,
+	tsexec stmgr.Executor,
+	syscalls vm.SyscallBuilder,
+	us stmgr.UpgradeSchedule,
+	_ dtypes.AfterGenesisSet) (dtypes.NetworkName, error) {
 	if !build.Devnet {
 		return "testnetnet", nil
 	}
 
 	ctx := helpers.LifecycleCtx(mctx, lc)
 
-	sm, err := stmgr.NewStateManagerWithUpgradeSchedule(cs, syscalls, us)
+	sm, err := stmgr.NewStateManager(cs, tsexec, syscalls, us, nil)
 	if err != nil {
 		return "", err
 	}
@@ -126,7 +139,8 @@ type SyncerParams struct {
 	SyncMgrCtor  chain.SyncManagerCtor
 	Host         host.Host
 	Beacon       beacon.Schedule
-	Verifier     ffiwrapper.Verifier
+	Gent         chain.Genesis
+	Consensus    consensus.Consensus
 }
 
 func NewSyncer(params SyncerParams) (*chain.Syncer, error) {
@@ -138,9 +152,8 @@ func NewSyncer(params SyncerParams) (*chain.Syncer, error) {
 		smCtor = params.SyncMgrCtor
 		h      = params.Host
 		b      = params.Beacon
-		v      = params.Verifier
 	)
-	syncer, err := chain.NewSyncer(ds, sm, ex, smCtor, h.ConnManager(), h.ID(), b, v)
+	syncer, err := chain.NewSyncer(ds, sm, ex, smCtor, h.ConnManager(), h.ID(), b, params.Gent, params.Consensus)
 	if err != nil {
 		return nil, err
 	}

node/modules/client.go

@@ -36,6 +36,7 @@ import (
 	"github.com/filecoin-project/lotus/markets"
 	marketevents "github.com/filecoin-project/lotus/markets/loggers"
 	"github.com/filecoin-project/lotus/markets/retrievaladapter"
+	"github.com/filecoin-project/lotus/node/config"
 	"github.com/filecoin-project/lotus/node/impl/full"
 	payapi "github.com/filecoin-project/lotus/node/impl/paych"
 	"github.com/filecoin-project/lotus/node/modules/dtypes"
@@ -112,7 +113,7 @@ func NewClientGraphsyncDataTransfer(lc fx.Lifecycle, h host.Host, gs dtypes.Grap
 	net := dtnet.NewFromLibp2pHost(h, dtRetryParams)
 
 	dtDs := namespace.Wrap(ds, datastore.NewKey("/datatransfer/client/transfers"))
-	transport := dtgstransport.NewTransport(h.ID(), gs)
+	transport := dtgstransport.NewTransport(h.ID(), gs, net)
 	err := os.MkdirAll(filepath.Join(r.Path(), "data-transfer"), 0755) //nolint: gosec
 	if err != nil && !os.IsExist(err) {
 		return nil, err
@@ -182,7 +183,7 @@ func StorageClient(lc fx.Lifecycle, h host.Host, dataTransfer dtypes.ClientDataT
 	marketsRetryParams := smnet.RetryParameters(time.Second, 5*time.Minute, 15, 5)
 	net := smnet.NewFromLibp2pHost(h, marketsRetryParams)
 
-	c, err := storageimpl.NewClient(net, dataTransfer, discovery, deals, scn, accessor, storageimpl.DealPollingInterval(time.Second))
+	c, err := storageimpl.NewClient(net, dataTransfer, discovery, deals, scn, accessor, storageimpl.DealPollingInterval(time.Second), storageimpl.MaxTraversalLinks(config.MaxTraversalLinks))
 	if err != nil {
 		return nil, err
 	}

node/modules/graphsync.go

@@ -1,30 +1,41 @@
 package modules
 
 import (
+	"context"
+	"time"
+
 	"github.com/ipfs/go-graphsync"
 	graphsyncimpl "github.com/ipfs/go-graphsync/impl"
 	gsnet "github.com/ipfs/go-graphsync/network"
 	"github.com/ipfs/go-graphsync/storeutil"
 	"github.com/libp2p/go-libp2p-core/host"
 	"github.com/libp2p/go-libp2p-core/peer"
+	"go.opencensus.io/stats"
 	"go.uber.org/fx"
 
+	"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"
 	"github.com/filecoin-project/lotus/node/repo"
 )
 
 // Graphsync creates a graphsync instance from the given loader and storer
-func Graphsync(parallelTransfers uint64) func(mctx helpers.MetricsCtx, lc fx.Lifecycle, r repo.LockedRepo, clientBs dtypes.ClientBlockstore, chainBs dtypes.ExposedBlockstore, h host.Host) (dtypes.Graphsync, error) {
+func Graphsync(parallelTransfersForStorage uint64, parallelTransfersForRetrieval uint64) func(mctx helpers.MetricsCtx, lc fx.Lifecycle, r repo.LockedRepo, clientBs dtypes.ClientBlockstore, chainBs dtypes.ExposedBlockstore, h host.Host) (dtypes.Graphsync, error) {
 	return func(mctx helpers.MetricsCtx, lc fx.Lifecycle, r repo.LockedRepo, clientBs dtypes.ClientBlockstore, chainBs dtypes.ExposedBlockstore, h host.Host) (dtypes.Graphsync, error) {
 		graphsyncNetwork := gsnet.NewFromLibp2pHost(h)
-		loader := storeutil.LoaderForBlockstore(clientBs)
-		storer := storeutil.StorerForBlockstore(clientBs)
+		lsys := storeutil.LinkSystemForBlockstore(clientBs)
 
-		gs := graphsyncimpl.New(helpers.LifecycleCtx(mctx, lc), graphsyncNetwork, loader, storer, graphsyncimpl.RejectAllRequestsByDefault(), graphsyncimpl.MaxInProgressRequests(parallelTransfers))
-		chainLoader := storeutil.LoaderForBlockstore(chainBs)
-		chainStorer := storeutil.StorerForBlockstore(chainBs)
-		err := gs.RegisterPersistenceOption("chainstore", chainLoader, chainStorer)
+		gs := graphsyncimpl.New(helpers.LifecycleCtx(mctx, lc),
+			graphsyncNetwork,
+			lsys,
+			graphsyncimpl.RejectAllRequestsByDefault(),
+			graphsyncimpl.MaxInProgressIncomingRequests(parallelTransfersForStorage),
+			graphsyncimpl.MaxInProgressOutgoingRequests(parallelTransfersForRetrieval),
+			graphsyncimpl.MaxLinksPerIncomingRequests(config.MaxTraversalLinks),
+			graphsyncimpl.MaxLinksPerOutgoingRequests(config.MaxTraversalLinks))
+		chainLinkSystem := storeutil.LinkSystemForBlockstore(chainBs)
+		err := gs.RegisterPersistenceOption("chainstore", chainLinkSystem)
 		if err != nil {
 			return nil, err
 		}
@@ -43,6 +54,48 @@ func Graphsync(parallelTransfers uint64) func(mctx helpers.MetricsCtx, lc fx.Lif
 				hookActions.UsePersistenceOption("chainstore")
 			}
 		})
+
+		graphsyncStats(mctx, lc, gs)
+
 		return gs, nil
 	}
 }
+
+func graphsyncStats(mctx helpers.MetricsCtx, lc fx.Lifecycle, gs dtypes.Graphsync) {
+	stopStats := make(chan struct{})
+	lc.Append(fx.Hook{
+		OnStart: func(context.Context) error {
+			go func() {
+				t := time.NewTicker(10 * time.Second)
+				for {
+					select {
+					case <-t.C:
+
+						st := gs.Stats()
+						stats.Record(mctx, metrics.GraphsyncReceivingPeersCount.M(int64(st.OutgoingRequests.TotalPeers)))
+						stats.Record(mctx, metrics.GraphsyncReceivingActiveCount.M(int64(st.OutgoingRequests.Active)))
+						stats.Record(mctx, metrics.GraphsyncReceivingCountCount.M(int64(st.OutgoingRequests.Pending)))
+						stats.Record(mctx, metrics.GraphsyncReceivingTotalMemoryAllocated.M(int64(st.IncomingResponses.TotalAllocatedAllPeers)))
+						stats.Record(mctx, metrics.GraphsyncReceivingTotalPendingAllocations.M(int64(st.IncomingResponses.TotalPendingAllocations)))
+						stats.Record(mctx, metrics.GraphsyncReceivingPeersPending.M(int64(st.IncomingResponses.NumPeersWithPendingAllocations)))
+						stats.Record(mctx, metrics.GraphsyncSendingPeersCount.M(int64(st.IncomingRequests.TotalPeers)))
+						stats.Record(mctx, metrics.GraphsyncSendingActiveCount.M(int64(st.IncomingRequests.Active)))
+						stats.Record(mctx, metrics.GraphsyncSendingCountCount.M(int64(st.IncomingRequests.Pending)))
+						stats.Record(mctx, metrics.GraphsyncSendingTotalMemoryAllocated.M(int64(st.OutgoingResponses.TotalAllocatedAllPeers)))
+						stats.Record(mctx, metrics.GraphsyncSendingTotalPendingAllocations.M(int64(st.OutgoingResponses.TotalPendingAllocations)))
+						stats.Record(mctx, metrics.GraphsyncSendingPeersPending.M(int64(st.OutgoingResponses.NumPeersWithPendingAllocations)))
+
+					case <-stopStats:
+						return
+					}
+				}
+			}()
+
+			return nil
+		},
+		OnStop: func(ctx context.Context) error {
+			close(stopStats)
+			return nil
+		},
+	})
+}

node/modules/lp2p/addrs.go

@@ -11,19 +11,6 @@ import (
 	mamask "github.com/whyrusleeping/multiaddr-filter"
 )
 
-func AddrFilters(filters []string) func() (opts Libp2pOpts, err error) {
-	return func() (opts Libp2pOpts, err error) {
-		for _, s := range filters {
-			f, err := mamask.NewMask(s)
-			if err != nil {
-				return opts, fmt.Errorf("incorrectly formatted address filter in config: %s", s)
-			}
-			opts.Opts = append(opts.Opts, libp2p.FilterAddresses(f)) //nolint:staticcheck
-		}
-		return opts, nil
-	}
-}
-
 func makeAddrsFactory(announce []string, noAnnounce []string) (p2pbhost.AddrsFactory, error) {
 	var annAddrs []ma.Multiaddr
 	for _, addr := range announce {

node/modules/lp2p/libp2p.go

@@ -42,7 +42,7 @@ func PrivKey(ks types.KeyStore) (crypto.PrivKey, error) {
 	if err != nil {
 		return nil, err
 	}
-	kbytes, err := pk.Bytes()
+	kbytes, err := crypto.MarshalPrivateKey(pk)
 	if err != nil {
 		return nil, err
 	}

node/modules/services.go

@@ -6,6 +6,7 @@ import (
 	"strconv"
 	"time"
 
+	"github.com/filecoin-project/lotus/chain/consensus"
 	"github.com/ipfs/go-datastore"
 	"github.com/ipfs/go-datastore/namespace"
 	eventbus "github.com/libp2p/go-eventbus"
@@ -30,6 +31,7 @@ import (
 	"github.com/filecoin-project/lotus/chain/sub"
 	"github.com/filecoin-project/lotus/chain/types"
 	"github.com/filecoin-project/lotus/journal"
+	"github.com/filecoin-project/lotus/journal/fsjournal"
 	"github.com/filecoin-project/lotus/lib/peermgr"
 	marketevents "github.com/filecoin-project/lotus/markets/loggers"
 	"github.com/filecoin-project/lotus/node/hello"
@@ -135,11 +137,19 @@ func waitForSync(stmgr *stmgr.StateManager, epochs int, subscribe func()) {
 	})
 }
 
-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) {
+func HandleIncomingBlocks(mctx helpers.MetricsCtx,
+	lc fx.Lifecycle,
+	ps *pubsub.PubSub,
+	s *chain.Syncer,
+	bserv dtypes.ChainBlockService,
+	chain *store.ChainStore,
+	cns consensus.Consensus,
+	h host.Host,
+	nn dtypes.NetworkName) {
 	ctx := helpers.LifecycleCtx(mctx, lc)
 
 	v := sub.NewBlockValidator(
-		h.ID(), chain, stmgr,
+		h.ID(), chain, cns,
 		func(p peer.ID) {
 			ps.BlacklistPeer(p)
 			h.ConnManager().TagPeer(p, "badblock", -1000)
@@ -237,7 +247,7 @@ func RandomSchedule(p RandomBeaconParams, _ dtypes.AfterGenesisSet) (beacon.Sche
 }
 
 func OpenFilesystemJournal(lr repo.LockedRepo, lc fx.Lifecycle, disabled journal.DisabledEvents) (journal.Journal, error) {
-	jrnl, err := journal.OpenFSJournal(lr, disabled)
+	jrnl, err := fsjournal.OpenFSJournal(lr, disabled)
 	if err != nil {
 		return nil, err
 	}

node/modules/stmgr.go

@@ -1,6 +1,7 @@
 package modules
 
 import (
+	"github.com/filecoin-project/lotus/chain/beacon"
 	"github.com/filecoin-project/lotus/chain/vm"
 	"go.uber.org/fx"
 
@@ -8,8 +9,8 @@ import (
 	"github.com/filecoin-project/lotus/chain/store"
 )
 
-func StateManager(lc fx.Lifecycle, cs *store.ChainStore, sys vm.SyscallBuilder, us stmgr.UpgradeSchedule) (*stmgr.StateManager, error) {
-	sm, err := stmgr.NewStateManagerWithUpgradeSchedule(cs, sys, us)
+func StateManager(lc fx.Lifecycle, cs *store.ChainStore, exec stmgr.Executor, sys vm.SyscallBuilder, us stmgr.UpgradeSchedule, b beacon.Schedule) (*stmgr.StateManager, error) {
+	sm, err := stmgr.NewStateManager(cs, exec, sys, us, b)
 	if err != nil {
 		return nil, err
 	}

node/modules/storageminer.go

@@ -38,6 +38,7 @@ import (
 	"github.com/ipfs/go-datastore"
 	"github.com/ipfs/go-datastore/namespace"
 	graphsync "github.com/ipfs/go-graphsync/impl"
+	graphsyncimpl "github.com/ipfs/go-graphsync/impl"
 	gsnet "github.com/ipfs/go-graphsync/network"
 	"github.com/ipfs/go-graphsync/storeutil"
 	"github.com/libp2p/go-libp2p-core/host"
@@ -70,7 +71,10 @@ import (
 	"github.com/filecoin-project/lotus/storage"
 )
 
-var StorageCounterDSPrefix = "/storage/nextid"
+var (
+	StorageCounterDSPrefix = "/storage/nextid"
+	StagingAreaDirName     = "deal-staging"
+)
 
 func minerAddrFromDS(ds dtypes.MetadataDS) (address.Address, error) {
 	maddrb, err := ds.Get(datastore.NewKey("miner-address"))
@@ -337,7 +341,7 @@ func NewProviderDAGServiceDataTransfer(lc fx.Lifecycle, h host.Host, gs dtypes.S
 	net := dtnet.NewFromLibp2pHost(h)
 
 	dtDs := namespace.Wrap(ds, datastore.NewKey("/datatransfer/provider/transfers"))
-	transport := dtgstransport.NewTransport(h.ID(), gs)
+	transport := dtgstransport.NewTransport(h.ID(), gs, net)
 	err := os.MkdirAll(filepath.Join(r.Path(), "data-transfer"), 0755) //nolint: gosec
 	if err != nil && !os.IsExist(err) {
 		return nil, err
@@ -391,12 +395,20 @@ func StagingBlockstore(lc fx.Lifecycle, mctx helpers.MetricsCtx, r repo.LockedRe
 
 // StagingGraphsync creates a graphsync instance which reads and writes blocks
 // to the StagingBlockstore
-func StagingGraphsync(parallelTransfers uint64) func(mctx helpers.MetricsCtx, lc fx.Lifecycle, ibs dtypes.StagingBlockstore, h host.Host) dtypes.StagingGraphsync {
+func StagingGraphsync(parallelTransfersForStorage uint64, parallelTransfersForRetrieval uint64) func(mctx helpers.MetricsCtx, lc fx.Lifecycle, ibs dtypes.StagingBlockstore, h host.Host) dtypes.StagingGraphsync {
 	return func(mctx helpers.MetricsCtx, lc fx.Lifecycle, ibs dtypes.StagingBlockstore, h host.Host) dtypes.StagingGraphsync {
 		graphsyncNetwork := gsnet.NewFromLibp2pHost(h)
-		loader := storeutil.LoaderForBlockstore(ibs)
-		storer := storeutil.StorerForBlockstore(ibs)
-		gs := graphsync.New(helpers.LifecycleCtx(mctx, lc), graphsyncNetwork, loader, storer, graphsync.RejectAllRequestsByDefault(), graphsync.MaxInProgressRequests(parallelTransfers))
+		lsys := storeutil.LinkSystemForBlockstore(ibs)
+		gs := graphsync.New(helpers.LifecycleCtx(mctx, lc),
+			graphsyncNetwork,
+			lsys,
+			graphsync.RejectAllRequestsByDefault(),
+			graphsync.MaxInProgressIncomingRequests(parallelTransfersForRetrieval),
+			graphsync.MaxInProgressOutgoingRequests(parallelTransfersForStorage),
+			graphsyncimpl.MaxLinksPerIncomingRequests(config.MaxTraversalLinks),
+			graphsyncimpl.MaxLinksPerOutgoingRequests(config.MaxTraversalLinks))
+
+		graphsyncStats(mctx, lc, gs)
 
 		return gs
 	}
@@ -442,14 +454,16 @@ func NewStorageAsk(ctx helpers.MetricsCtx, fapi v1api.FullNode, ds dtypes.Metada
 		storagemarket.MaxPieceSize(abi.PaddedPieceSize(mi.SectorSize)))
 }
 
-func BasicDealFilter(user dtypes.StorageDealFilter) func(onlineOk dtypes.ConsiderOnlineStorageDealsConfigFunc,
+func BasicDealFilter(cfg config.DealmakingConfig, user dtypes.StorageDealFilter) func(onlineOk dtypes.ConsiderOnlineStorageDealsConfigFunc,
 	offlineOk dtypes.ConsiderOfflineStorageDealsConfigFunc,
 	verifiedOk dtypes.ConsiderVerifiedStorageDealsConfigFunc,
 	unverifiedOk dtypes.ConsiderUnverifiedStorageDealsConfigFunc,
 	blocklistFunc dtypes.StorageDealPieceCidBlocklistConfigFunc,
 	expectedSealTimeFunc dtypes.GetExpectedSealDurationFunc,
 	startDelay dtypes.GetMaxDealStartDelayFunc,
-	spn storagemarket.StorageProviderNode) dtypes.StorageDealFilter {
+	spn storagemarket.StorageProviderNode,
+	r repo.LockedRepo,
+) dtypes.StorageDealFilter {
 	return func(onlineOk dtypes.ConsiderOnlineStorageDealsConfigFunc,
 		offlineOk dtypes.ConsiderOfflineStorageDealsConfigFunc,
 		verifiedOk dtypes.ConsiderVerifiedStorageDealsConfigFunc,
@@ -457,7 +471,9 @@ func BasicDealFilter(user dtypes.StorageDealFilter) func(onlineOk dtypes.Conside
 		blocklistFunc dtypes.StorageDealPieceCidBlocklistConfigFunc,
 		expectedSealTimeFunc dtypes.GetExpectedSealDurationFunc,
 		startDelay dtypes.GetMaxDealStartDelayFunc,
-		spn storagemarket.StorageProviderNode) dtypes.StorageDealFilter {
+		spn storagemarket.StorageProviderNode,
+		r repo.LockedRepo,
+	) dtypes.StorageDealFilter {
 
 		return func(ctx context.Context, deal storagemarket.MinerDeal) (bool, string, error) {
 			b, err := onlineOk()
@@ -533,6 +549,17 @@ func BasicDealFilter(user dtypes.StorageDealFilter) func(onlineOk dtypes.Conside
 				return false, "miner error", err
 			}
 
+			dir := filepath.Join(r.Path(), StagingAreaDirName)
+			diskUsageBytes, err := r.DiskUsage(dir)
+			if err != nil {
+				return false, "miner error", err
+			}
+
+			if cfg.MaxStagingDealsBytes != 0 && diskUsageBytes >= cfg.MaxStagingDealsBytes {
+				log.Errorw("proposed deal rejected because there are too many deals in the staging area at the moment", "MaxStagingDealsBytes", cfg.MaxStagingDealsBytes, "DiskUsageBytes", diskUsageBytes)
+				return false, "cannot accept deal as miner is overloaded at the moment - there are too many staging deals being processed", nil
+			}
+
 			// Reject if it's more than 7 days in the future
 			// TODO: read from cfg
 			maxStartEpoch := earliest + abi.ChainEpoch(uint64(sd.Seconds())/build.BlockDelaySecs)
@@ -561,7 +588,7 @@ func StorageProvider(minerAddress dtypes.MinerAddress,
 ) (storagemarket.StorageProvider, error) {
 	net := smnet.NewFromLibp2pHost(h)
 
-	dir := filepath.Join(r.Path(), "deal-staging")
+	dir := filepath.Join(r.Path(), StagingAreaDirName)
 
 	// migrate temporary files that were created directly under the repo, by
 	// moving them to the new directory and symlinking them.
@@ -853,12 +880,13 @@ 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,
-				MaxSealingSectorsForDeals: cfg.MaxSealingSectorsForDeals,
-				WaitDealsDelay:            config.Duration(cfg.WaitDealsDelay),
-				AlwaysKeepUnsealedCopy:    cfg.AlwaysKeepUnsealedCopy,
-				FinalizeEarly:             cfg.FinalizeEarly,
+				MaxWaitDealsSectors:             cfg.MaxWaitDealsSectors,
+				MaxSealingSectors:               cfg.MaxSealingSectors,
+				MaxSealingSectorsForDeals:       cfg.MaxSealingSectorsForDeals,
+				CommittedCapacitySectorLifetime: config.Duration(cfg.CommittedCapacitySectorLifetime),
+				WaitDealsDelay:                  config.Duration(cfg.WaitDealsDelay),
+				AlwaysKeepUnsealedCopy:          cfg.AlwaysKeepUnsealedCopy,
+				FinalizeEarly:                   cfg.FinalizeEarly,
 
 				CollateralFromMinerBalance: cfg.CollateralFromMinerBalance,
 				AvailableBalanceBuffer:     types.FIL(cfg.AvailableBalanceBuffer),
@@ -869,12 +897,13 @@ func NewSetSealConfigFunc(r repo.LockedRepo) (dtypes.SetSealingConfigFunc, error
 				PreCommitBatchWait:  config.Duration(cfg.PreCommitBatchWait),
 				PreCommitBatchSlack: config.Duration(cfg.PreCommitBatchSlack),
 
-				AggregateCommits:      cfg.AggregateCommits,
-				MinCommitBatch:        cfg.MinCommitBatch,
-				MaxCommitBatch:        cfg.MaxCommitBatch,
-				CommitBatchWait:       config.Duration(cfg.CommitBatchWait),
-				CommitBatchSlack:      config.Duration(cfg.CommitBatchSlack),
-				AggregateAboveBaseFee: types.FIL(cfg.AggregateAboveBaseFee),
+				AggregateCommits:           cfg.AggregateCommits,
+				MinCommitBatch:             cfg.MinCommitBatch,
+				MaxCommitBatch:             cfg.MaxCommitBatch,
+				CommitBatchWait:            config.Duration(cfg.CommitBatchWait),
+				CommitBatchSlack:           config.Duration(cfg.CommitBatchSlack),
+				AggregateAboveBaseFee:      types.FIL(cfg.AggregateAboveBaseFee),
+				BatchPreCommitAboveBaseFee: types.FIL(cfg.BatchPreCommitAboveBaseFee),
 
 				TerminateBatchMax:  cfg.TerminateBatchMax,
 				TerminateBatchMin:  cfg.TerminateBatchMin,
@@ -887,12 +916,13 @@ func NewSetSealConfigFunc(r repo.LockedRepo) (dtypes.SetSealingConfigFunc, error
 
 func ToSealingConfig(cfg *config.StorageMiner) sealiface.Config {
 	return sealiface.Config{
-		MaxWaitDealsSectors:       cfg.Sealing.MaxWaitDealsSectors,
-		MaxSealingSectors:         cfg.Sealing.MaxSealingSectors,
-		MaxSealingSectorsForDeals: cfg.Sealing.MaxSealingSectorsForDeals,
-		WaitDealsDelay:            time.Duration(cfg.Sealing.WaitDealsDelay),
-		AlwaysKeepUnsealedCopy:    cfg.Sealing.AlwaysKeepUnsealedCopy,
-		FinalizeEarly:             cfg.Sealing.FinalizeEarly,
+		MaxWaitDealsSectors:             cfg.Sealing.MaxWaitDealsSectors,
+		MaxSealingSectors:               cfg.Sealing.MaxSealingSectors,
+		MaxSealingSectorsForDeals:       cfg.Sealing.MaxSealingSectorsForDeals,
+		CommittedCapacitySectorLifetime: time.Duration(cfg.Sealing.CommittedCapacitySectorLifetime),
+		WaitDealsDelay:                  time.Duration(cfg.Sealing.WaitDealsDelay),
+		AlwaysKeepUnsealedCopy:          cfg.Sealing.AlwaysKeepUnsealedCopy,
+		FinalizeEarly:                   cfg.Sealing.FinalizeEarly,
 
 		CollateralFromMinerBalance: cfg.Sealing.CollateralFromMinerBalance,
 		AvailableBalanceBuffer:     types.BigInt(cfg.Sealing.AvailableBalanceBuffer),
@@ -903,16 +933,19 @@ func ToSealingConfig(cfg *config.StorageMiner) sealiface.Config {
 		PreCommitBatchWait:  time.Duration(cfg.Sealing.PreCommitBatchWait),
 		PreCommitBatchSlack: time.Duration(cfg.Sealing.PreCommitBatchSlack),
 
-		AggregateCommits:      cfg.Sealing.AggregateCommits,
-		MinCommitBatch:        cfg.Sealing.MinCommitBatch,
-		MaxCommitBatch:        cfg.Sealing.MaxCommitBatch,
-		CommitBatchWait:       time.Duration(cfg.Sealing.CommitBatchWait),
-		CommitBatchSlack:      time.Duration(cfg.Sealing.CommitBatchSlack),
-		AggregateAboveBaseFee: types.BigInt(cfg.Sealing.AggregateAboveBaseFee),
+		AggregateCommits:           cfg.Sealing.AggregateCommits,
+		MinCommitBatch:             cfg.Sealing.MinCommitBatch,
+		MaxCommitBatch:             cfg.Sealing.MaxCommitBatch,
+		CommitBatchWait:            time.Duration(cfg.Sealing.CommitBatchWait),
+		CommitBatchSlack:           time.Duration(cfg.Sealing.CommitBatchSlack),
+		AggregateAboveBaseFee:      types.BigInt(cfg.Sealing.AggregateAboveBaseFee),
+		BatchPreCommitAboveBaseFee: types.BigInt(cfg.Sealing.BatchPreCommitAboveBaseFee),
 
 		TerminateBatchMax:  cfg.Sealing.TerminateBatchMax,
 		TerminateBatchMin:  cfg.Sealing.TerminateBatchMin,
 		TerminateBatchWait: time.Duration(cfg.Sealing.TerminateBatchWait),
+
+		StartEpochSealingBuffer: abi.ChainEpoch(cfg.Dealmaking.StartEpochSealingBuffer),
 	}
 }