Merge pull request #3531 from filecoin-project/asr/changelog

Lotus version 0.5.10

.codecov.yml

@@ -1,3 +1,5 @@
 comment: off
 ignore:
   - "cbor_gen.go"
+github_checks:
+  annotations: false

.github/labels.yml

@@ -26,11 +26,14 @@
   color: 00A4E2
   description: "Area: Chain/VM"
 - name: area/chain/sync
-  color: 00A4E2
+  color: 00A4E4
   description: "Area: Chain/Sync"
 - name: area/chain/misc
-  color: 00A4E2
+  color: 00A4E6
   description: "Area: Chain/Misc"
+- name: area/markets
+  color: 00A4E8
+  description: "Area: Markets"  
 - name: area/sealing/fsm
   color: 0bb1ed
   description: "Area: Sealing/FSM"
@@ -149,6 +152,9 @@
 - name: impact/test-flakiness
   color: DDE1E4
   description: "Impact: Test Flakiness"
+- name: impact/consensus
+  color: b20014
+  description: "Impact: Consensus"
 
 ###
 ### Topics

CHANGELOG.md

@@ -1,5 +1,60 @@
 # Lotus changelog
 
+# 0.5.10 / 2020-09-03
+
+This patch includes a crucial fix to the message pool selection logic, strongly disfavouring messages that might cause a miner penalty.
+
+## Changes
+
+- Fix calculation of GasReward in messagepool (https://github.com/filecoin-project/lotus/pull/3528)
+
+# 0.5.9 / 2020-09-03
+
+This patch includes a hotfix to the `GasEstimateFeeCap` method, capping the estimated fee to a reasonable level by default.
+
+## Changes 
+
+- Added target height to sync wait (https://github.com/filecoin-project/lotus/pull/3502)
+- Disable codecov annotations (https://github.com/filecoin-project/lotus/pull/3514)
+- Cap fees to reasonable level by default (https://github.com/filecoin-project/lotus/pull/3516)
+- Add APIs and command to inspect bandwidth usage (https://github.com/filecoin-project/lotus/pull/3497)
+- Track expected nonce in mpool, ignore messages with large nonce gaps (https://github.com/filecoin-project/lotus/pull/3450)
+
+# 0.5.8 / 2020-09-02
+
+This patch includes some bugfixes to the sector sealing process, and updates go-fil-markets. It also improves the performance of blocksync, adds a method to export chain state trees, and improves chainwatch.
+
+## Changes
+
+- Upgrade markets to v0.5.9 (https://github.com/filecoin-project/lotus/pull/3496)
+- Improve blocksync to load fewer messages: (https://github.com/filecoin-project/lotus/pull/3494)
+- Fix a panic in the ffi-wrapper's `ReadPiece` (https://github.com/filecoin-project/lotus/pull/3492/files)
+- Fix a deadlock in the sealing scheduler (https://github.com/filecoin-project/lotus/pull/3489)
+- Add test vectors for tipset tests (https://github.com/filecoin-project/lotus/pull/3485/files)
+- Improve the advance-block debug command (https://github.com/filecoin-project/lotus/pull/3476)
+- Add toggle for message processing to Lotus PCR (https://github.com/filecoin-project/lotus/pull/3470)
+- Allow exporting recent chain state trees (https://github.com/filecoin-project/lotus/pull/3463)
+- Remove height from chain rand (https://github.com/filecoin-project/lotus/pull/3458)
+- Disable GC on chain badger datastore (https://github.com/filecoin-project/lotus/pull/3457)
+- Account for `GasPremium` in `GasEstimateFeeCap` (https://github.com/filecoin-project/lotus/pull/3456)
+- Update go-libp2p-pubsub to `master` (https://github.com/filecoin-project/lotus/pull/3455)
+- Chainwatch improvements (https://github.com/filecoin-project/lotus/pull/3442)
+
+# 0.5.7 / 2020-08-31
+
+This patch release includes some bugfixes and enhancements to the sector lifecycle and message pool logic. 
+
+## Changes
+
+- Rebuild unsealed infos on miner restart (https://github.com/filecoin-project/lotus/pull/3401)
+- CLI to attach storage paths to workers (https://github.com/filecoin-project/lotus/pull/3405)
+- Do not select negative performing message chains for inclusion (https://github.com/filecoin-project/lotus/pull/3392)
+- Remove a redundant error-check (https://github.com/filecoin-project/lotus/pull/3421)
+- Correctly move unsealed sectors in `FinalizeSectors` (https://github.com/filecoin-project/lotus/pull/3424)
+- Improve worker selection logic (https://github.com/filecoin-project/lotus/pull/3425)
+- Don't use context to close bitswap (https://github.com/filecoin-project/lotus/pull/3430)
+- Correctly estimate gas premium when there is only one message on chain (https://github.com/filecoin-project/lotus/pull/3428)
+
 # 0.5.6 / 2020-08-29
 
 Hotfix release that fixes a panic in the sealing scheduler (https://github.com/filecoin-project/lotus/pull/3389).

api/api_common.go

@@ -5,8 +5,10 @@ import (
 	"fmt"
 
 	"github.com/filecoin-project/go-jsonrpc/auth"
+	metrics "github.com/libp2p/go-libp2p-core/metrics"
 	"github.com/libp2p/go-libp2p-core/network"
 	"github.com/libp2p/go-libp2p-core/peer"
+	protocol "github.com/libp2p/go-libp2p-core/protocol"
 
 	"github.com/filecoin-project/lotus/build"
 )
@@ -29,6 +31,18 @@ type Common interface {
 	NetPubsubScores(context.Context) ([]PubsubScore, error)
 	NetAutoNatStatus(context.Context) (NatInfo, error)
 
+	// NetBandwidthStats returns statistics about the nodes total bandwidth
+	// usage and current rate across all peers and protocols.
+	NetBandwidthStats(ctx context.Context) (metrics.Stats, error)
+
+	// NetBandwidthStatsByPeer returns statistics about the nodes bandwidth
+	// usage and current rate per peer
+	NetBandwidthStatsByPeer(ctx context.Context) (map[string]metrics.Stats, error)
+
+	// NetBandwidthStatsByProtocol returns statistics about the nodes bandwidth
+	// usage and current rate per protocol
+	NetBandwidthStatsByProtocol(ctx context.Context) (map[protocol.ID]metrics.Stats, error)
+
 	// MethodGroup: Common
 
 	// ID returns peerID of libp2p node backing this API

api/api_full.go

@@ -109,7 +109,10 @@ type FullNode interface {
 	ChainGetPath(ctx context.Context, from types.TipSetKey, to types.TipSetKey) ([]*HeadChange, error)
 
 	// ChainExport returns a stream of bytes with CAR dump of chain data.
-	ChainExport(context.Context, types.TipSetKey) (<-chan []byte, error)
+	// The exported chain data includes the header chain from the given tipset
+	// back to genesis, the entire genesis state, and the most recent 'nroots'
+	// state trees.
+	ChainExport(ctx context.Context, nroots abi.ChainEpoch, tsk types.TipSetKey) (<-chan []byte, error)
 
 	// MethodGroup: Beacon
 	// The Beacon method group contains methods for interacting with the random beacon (DRAND)

api/api_worker.go

@@ -27,11 +27,13 @@ type WorkerAPI interface {
 
 	storage.Sealer
 
-	MoveStorage(ctx context.Context, sector abi.SectorID) error
+	MoveStorage(ctx context.Context, sector abi.SectorID, types stores.SectorFileType) error
 
 	UnsealPiece(context.Context, abi.SectorID, storiface.UnpaddedByteIndex, abi.UnpaddedPieceSize, abi.SealRandomness, cid.Cid) error
 	ReadPiece(context.Context, io.Writer, abi.SectorID, storiface.UnpaddedByteIndex, abi.UnpaddedPieceSize) (bool, error)
 
+	StorageAddLocal(ctx context.Context, path string) error
+
 	Fetch(context.Context, abi.SectorID, stores.SectorFileType, stores.PathType, stores.AcquireMode) error
 
 	Closing(context.Context) (<-chan struct{}, error)

api/apistruct/struct.go

@@ -6,8 +6,10 @@ import (
 	"time"
 
 	"github.com/ipfs/go-cid"
+	metrics "github.com/libp2p/go-libp2p-core/metrics"
 	"github.com/libp2p/go-libp2p-core/network"
 	"github.com/libp2p/go-libp2p-core/peer"
+	protocol "github.com/libp2p/go-libp2p-core/protocol"
 
 	"github.com/filecoin-project/go-address"
 	"github.com/filecoin-project/go-fil-markets/piecestore"
@@ -42,14 +44,17 @@ type CommonStruct struct {
 		AuthVerify func(ctx context.Context, token string) ([]auth.Permission, error) `perm:"read"`
 		AuthNew    func(ctx context.Context, perms []auth.Permission) ([]byte, error) `perm:"admin"`
 
-		NetConnectedness func(context.Context, peer.ID) (network.Connectedness, error) `perm:"read"`
-		NetPeers         func(context.Context) ([]peer.AddrInfo, error)                `perm:"read"`
-		NetConnect       func(context.Context, peer.AddrInfo) error                    `perm:"write"`
-		NetAddrsListen   func(context.Context) (peer.AddrInfo, error)                  `perm:"read"`
-		NetDisconnect    func(context.Context, peer.ID) error                          `perm:"write"`
-		NetFindPeer      func(context.Context, peer.ID) (peer.AddrInfo, error)         `perm:"read"`
-		NetPubsubScores  func(context.Context) ([]api.PubsubScore, error)              `perm:"read"`
-		NetAutoNatStatus func(context.Context) (api.NatInfo, error)                    `perm:"read"`
+		NetConnectedness            func(context.Context, peer.ID) (network.Connectedness, error)    `perm:"read"`
+		NetPeers                    func(context.Context) ([]peer.AddrInfo, error)                   `perm:"read"`
+		NetConnect                  func(context.Context, peer.AddrInfo) error                       `perm:"write"`
+		NetAddrsListen              func(context.Context) (peer.AddrInfo, error)                     `perm:"read"`
+		NetDisconnect               func(context.Context, peer.ID) error                             `perm:"write"`
+		NetFindPeer                 func(context.Context, peer.ID) (peer.AddrInfo, error)            `perm:"read"`
+		NetPubsubScores             func(context.Context) ([]api.PubsubScore, error)                 `perm:"read"`
+		NetAutoNatStatus            func(context.Context) (api.NatInfo, error)                       `perm:"read"`
+		NetBandwidthStats           func(ctx context.Context) (metrics.Stats, error)                 `perm:"read"`
+		NetBandwidthStatsByPeer     func(ctx context.Context) (map[string]metrics.Stats, error)      `perm:"read"`
+		NetBandwidthStatsByProtocol func(ctx context.Context) (map[protocol.ID]metrics.Stats, error) `perm:"read"`
 
 		ID      func(context.Context) (peer.ID, error)     `perm:"read"`
 		Version func(context.Context) (api.Version, error) `perm:"read"`
@@ -86,7 +91,7 @@ type FullNodeStruct struct {
 		ChainGetNode                  func(ctx context.Context, p string) (*api.IpldObject, error)                                                       `perm:"read"`
 		ChainGetMessage               func(context.Context, cid.Cid) (*types.Message, error)                                                             `perm:"read"`
 		ChainGetPath                  func(context.Context, types.TipSetKey, types.TipSetKey) ([]*api.HeadChange, error)                                 `perm:"read"`
-		ChainExport                   func(context.Context, types.TipSetKey) (<-chan []byte, error)                                                      `perm:"read"`
+		ChainExport                   func(context.Context, abi.ChainEpoch, types.TipSetKey) (<-chan []byte, error)                                      `perm:"read"`
 
 		BeaconGetEntry func(ctx context.Context, epoch abi.ChainEpoch) (*types.BeaconEntry, error) `perm:"read"`
 
@@ -317,7 +322,8 @@ type WorkerStruct struct {
 		FinalizeSector  func(context.Context, abi.SectorID, []storage.Range) error                                                                                                                                 `perm:"admin"`
 		ReleaseUnsealed func(ctx context.Context, sector abi.SectorID, safeToFree []storage.Range) error                                                                                                           `perm:"admin"`
 		Remove          func(ctx context.Context, sector abi.SectorID) error                                                                                                                                       `perm:"admin"`
-		MoveStorage     func(ctx context.Context, sector abi.SectorID) error                                                                                                                                       `perm:"admin"`
+		MoveStorage     func(ctx context.Context, sector abi.SectorID, types stores.SectorFileType) error                                                                                                          `perm:"admin"`
+		StorageAddLocal func(ctx context.Context, path string) error                                                                                                                                               `perm:"admin"`
 
 		UnsealPiece func(context.Context, abi.SectorID, storiface.UnpaddedByteIndex, abi.UnpaddedPieceSize, abi.SealRandomness, cid.Cid) error `perm:"admin"`
 		ReadPiece   func(context.Context, io.Writer, abi.SectorID, storiface.UnpaddedByteIndex, abi.UnpaddedPieceSize) (bool, error)           `perm:"admin"`
@@ -370,6 +376,18 @@ func (c *CommonStruct) NetAutoNatStatus(ctx context.Context) (api.NatInfo, error
 	return c.Internal.NetAutoNatStatus(ctx)
 }
 
+func (c *CommonStruct) NetBandwidthStats(ctx context.Context) (metrics.Stats, error) {
+	return c.Internal.NetBandwidthStats(ctx)
+}
+
+func (c *CommonStruct) NetBandwidthStatsByPeer(ctx context.Context) (map[string]metrics.Stats, error) {
+	return c.Internal.NetBandwidthStatsByPeer(ctx)
+}
+
+func (c *CommonStruct) NetBandwidthStatsByProtocol(ctx context.Context) (map[protocol.ID]metrics.Stats, error) {
+	return c.Internal.NetBandwidthStatsByProtocol(ctx)
+}
+
 // ID implements API.ID
 func (c *CommonStruct) ID(ctx context.Context) (peer.ID, error) {
 	return c.Internal.ID(ctx)
@@ -653,8 +671,8 @@ func (c *FullNodeStruct) ChainGetPath(ctx context.Context, from types.TipSetKey,
 	return c.Internal.ChainGetPath(ctx, from, to)
 }
 
-func (c *FullNodeStruct) ChainExport(ctx context.Context, tsk types.TipSetKey) (<-chan []byte, error) {
-	return c.Internal.ChainExport(ctx, tsk)
+func (c *FullNodeStruct) ChainExport(ctx context.Context, nroots abi.ChainEpoch, tsk types.TipSetKey) (<-chan []byte, error) {
+	return c.Internal.ChainExport(ctx, nroots, tsk)
 }
 
 func (c *FullNodeStruct) BeaconGetEntry(ctx context.Context, epoch abi.ChainEpoch) (*types.BeaconEntry, error) {
@@ -1219,8 +1237,12 @@ func (w *WorkerStruct) Remove(ctx context.Context, sector abi.SectorID) error {
 	return w.Internal.Remove(ctx, sector)
 }
 
-func (w *WorkerStruct) MoveStorage(ctx context.Context, sector abi.SectorID) error {
-	return w.Internal.MoveStorage(ctx, sector)
+func (w *WorkerStruct) MoveStorage(ctx context.Context, sector abi.SectorID, types stores.SectorFileType) error {
+	return w.Internal.MoveStorage(ctx, sector, types)
+}
+
+func (w *WorkerStruct) StorageAddLocal(ctx context.Context, path string) error {
+	return w.Internal.StorageAddLocal(ctx, path)
 }
 
 func (w *WorkerStruct) UnsealPiece(ctx context.Context, id abi.SectorID, index storiface.UnpaddedByteIndex, size abi.UnpaddedPieceSize, randomness abi.SealRandomness, c cid.Cid) error {

api/docgen/docgen.go

@@ -14,8 +14,10 @@ import (
 
 	"github.com/ipfs/go-cid"
 	"github.com/ipfs/go-filestore"
+	metrics "github.com/libp2p/go-libp2p-core/metrics"
 	"github.com/libp2p/go-libp2p-core/network"
 	"github.com/libp2p/go-libp2p-core/peer"
+	protocol "github.com/libp2p/go-libp2p-core/protocol"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	"github.com/multiformats/go-multiaddr"
 
@@ -133,6 +135,22 @@ func init() {
 			InvalidMessageDeliveries: 3,
 		},
 	})
+	addExample(map[string]metrics.Stats{
+		"12D3KooWSXmXLJmBR1M7i9RW9GQPNUhZSzXKzxDHWtAgNuJAbyEJ": {
+			RateIn:   100,
+			RateOut:  50,
+			TotalIn:  174000,
+			TotalOut: 12500,
+		},
+	})
+	addExample(map[protocol.ID]metrics.Stats{
+		"/fil/hello/1.0.0": {
+			RateIn:   100,
+			RateOut:  50,
+			TotalIn:  174000,
+			TotalOut: 12500,
+		},
+	})
 
 	maddr, err := multiaddr.NewMultiaddr("/ip4/52.36.61.156/tcp/1347/p2p/12D3KooWFETiESTf1v4PGUvtnxMAcEFMzLZbJGg4tjWfGEimYior")
 	if err != nil {

build/version.go

@@ -25,7 +25,7 @@ func buildType() string {
 }
 
 // BuildVersion is the local build version, set by build system
-const BuildVersion = "0.5.6"
+const BuildVersion = "0.5.10"
 
 func UserVersion() string {
 	return BuildVersion + buildType() + CurrentCommit
@@ -53,7 +53,7 @@ func (ve Version) EqMajorMinor(v2 Version) bool {
 }
 
 // APIVersion is a semver version of the rpc api exposed
-var APIVersion Version = newVer(0, 13, 0)
+var APIVersion Version = newVer(0, 14, 0)
 
 //nolint:varcheck,deadcode
 const (

chain/blocksync/server.go

@@ -221,37 +221,36 @@ func collectChainSegment(
 func gatherMessages(cs *store.ChainStore, ts *types.TipSet) ([]*types.Message, [][]uint64, []*types.SignedMessage, [][]uint64, error) {
 	blsmsgmap := make(map[cid.Cid]uint64)
 	secpkmsgmap := make(map[cid.Cid]uint64)
-	var secpkmsgs []*types.SignedMessage
-	var blsmsgs []*types.Message
 	var secpkincl, blsincl [][]uint64
 
+	var blscids, secpkcids []cid.Cid
 	for _, block := range ts.Blocks() {
-		bmsgs, smsgs, err := cs.MessagesForBlock(block)
+		bc, sc, err := cs.ReadMsgMetaCids(block.Messages)
 		if err != nil {
 			return nil, nil, nil, nil, err
 		}
 
 		// FIXME: DRY. Use `chain.Message` interface.
-		bmi := make([]uint64, 0, len(bmsgs))
-		for _, m := range bmsgs {
-			i, ok := blsmsgmap[m.Cid()]
+		bmi := make([]uint64, 0, len(bc))
+		for _, m := range bc {
+			i, ok := blsmsgmap[m]
 			if !ok {
-				i = uint64(len(blsmsgs))
-				blsmsgs = append(blsmsgs, m)
-				blsmsgmap[m.Cid()] = i
+				i = uint64(len(blscids))
+				blscids = append(blscids, m)
+				blsmsgmap[m] = i
 			}
 
 			bmi = append(bmi, i)
 		}
 		blsincl = append(blsincl, bmi)
 
-		smi := make([]uint64, 0, len(smsgs))
-		for _, m := range smsgs {
-			i, ok := secpkmsgmap[m.Cid()]
+		smi := make([]uint64, 0, len(sc))
+		for _, m := range sc {
+			i, ok := secpkmsgmap[m]
 			if !ok {
-				i = uint64(len(secpkmsgs))
-				secpkmsgs = append(secpkmsgs, m)
-				secpkmsgmap[m.Cid()] = i
+				i = uint64(len(secpkcids))
+				secpkcids = append(secpkcids, m)
+				secpkmsgmap[m] = i
 			}
 
 			smi = append(smi, i)
@@ -259,5 +258,15 @@ func gatherMessages(cs *store.ChainStore, ts *types.TipSet) ([]*types.Message, [
 		secpkincl = append(secpkincl, smi)
 	}
 
+	blsmsgs, err := cs.LoadMessagesFromCids(blscids)
+	if err != nil {
+		return nil, nil, nil, nil, err
+	}
+
+	secpkmsgs, err := cs.LoadSignedMessagesFromCids(secpkcids)
+	if err != nil {
+		return nil, nil, nil, nil, err
+	}
+
 	return blsmsgs, blsincl, secpkmsgs, secpkincl, nil
 }

chain/messagepool/messagepool.go

@@ -52,6 +52,8 @@ var minimumBaseFee = types.NewInt(uint64(build.MinimumBaseFee))
 
 var MaxActorPendingMessages = 1000
 
+var MaxNonceGap = uint64(4)
+
 var (
 	ErrMessageTooBig = errors.New("message too big")
 
@@ -68,6 +70,7 @@ var (
 	ErrSoftValidationFailure  = errors.New("validation failure")
 	ErrRBFTooLowPremium       = errors.New("replace by fee has too low GasPremium")
 	ErrTooManyPendingMessages = errors.New("too many pending messages for actor")
+	ErrNonceGap               = errors.New("unfulfilled nonce gap")
 
 	ErrTryAgain = errors.New("state inconsistency while pushing message; please try again")
 )
@@ -131,19 +134,39 @@ type msgSet struct {
 	requiredFunds *stdbig.Int
 }
 
-func newMsgSet() *msgSet {
+func newMsgSet(nonce uint64) *msgSet {
 	return &msgSet{
 		msgs:          make(map[uint64]*types.SignedMessage),
+		nextNonce:     nonce,
 		requiredFunds: stdbig.NewInt(0),
 	}
 }
 
-func (ms *msgSet) add(m *types.SignedMessage, mp *MessagePool, limit bool) (bool, error) {
-	if len(ms.msgs) == 0 || m.Message.Nonce >= ms.nextNonce {
-		ms.nextNonce = m.Message.Nonce + 1
+func (ms *msgSet) add(m *types.SignedMessage, mp *MessagePool, strict bool) (bool, error) {
+	nextNonce := ms.nextNonce
+	nonceGap := false
+	switch {
+	case m.Message.Nonce == nextNonce:
+		nextNonce++
+		// advance if we are filling a gap
+		for _, fillGap := ms.msgs[nextNonce]; fillGap; _, fillGap = ms.msgs[nextNonce] {
+			nextNonce++
+		}
+
+	case strict && m.Message.Nonce > nextNonce+MaxNonceGap:
+		return false, xerrors.Errorf("message nonce has too big a gap from expected nonce (Nonce: %d, nextNonce: %d): %w", m.Message.Nonce, nextNonce, ErrNonceGap)
+
+	case m.Message.Nonce > nextNonce:
+		nonceGap = true
 	}
+
 	exms, has := ms.msgs[m.Message.Nonce]
 	if has {
+		// refuse RBF if we have a gap
+		if strict && nonceGap {
+			return false, xerrors.Errorf("rejecting replace by fee because of nonce gap (Nonce: %d, nextNonce: %d): %w", m.Message.Nonce, nextNonce, ErrNonceGap)
+		}
+
 		if m.Cid() != exms.Cid() {
 			// check if RBF passes
 			minPrice := exms.Message.GasPremium
@@ -159,17 +182,26 @@ func (ms *msgSet) add(m *types.SignedMessage, mp *MessagePool, limit bool) (bool
 					m.Message.From, m.Message.Nonce, minPrice, m.Message.GasPremium,
 					ErrRBFTooLowPremium)
 			}
+		} else {
+			return false, xerrors.Errorf("message from %s with nonce %d already in mpool: %w",
+				m.Message.From, m.Message.Nonce, ErrSoftValidationFailure)
 		}
 
 		ms.requiredFunds.Sub(ms.requiredFunds, exms.Message.RequiredFunds().Int)
 		//ms.requiredFunds.Sub(ms.requiredFunds, exms.Message.Value.Int)
 	}
 
-	if !has && limit && len(ms.msgs) > MaxActorPendingMessages {
+	if !has && strict && len(ms.msgs) > MaxActorPendingMessages {
 		log.Errorf("too many pending messages from actor %s", m.Message.From)
 		return false, ErrTooManyPendingMessages
 	}
 
+	if strict && nonceGap {
+		log.Warnf("adding nonce-gapped message from %s (nonce: %d, nextNonce: %d)",
+			m.Message.From, m.Message.Nonce, nextNonce)
+	}
+
+	ms.nextNonce = nextNonce
 	ms.msgs[m.Message.Nonce] = m
 	ms.requiredFunds.Add(ms.requiredFunds, m.Message.RequiredFunds().Int)
 	//ms.requiredFunds.Add(ms.requiredFunds, m.Message.Value.Int)
@@ -177,12 +209,38 @@ func (ms *msgSet) add(m *types.SignedMessage, mp *MessagePool, limit bool) (bool
 	return !has, nil
 }
 
-func (ms *msgSet) rm(nonce uint64) {
+func (ms *msgSet) rm(nonce uint64, applied bool) {
 	m, has := ms.msgs[nonce]
-	if has {
-		ms.requiredFunds.Sub(ms.requiredFunds, m.Message.RequiredFunds().Int)
-		//ms.requiredFunds.Sub(ms.requiredFunds, m.Message.Value.Int)
-		delete(ms.msgs, nonce)
+	if !has {
+		if applied && nonce >= ms.nextNonce {
+			// we removed a message we did not know about because it was applied
+			// we need to adjust the nonce and check if we filled a gap
+			ms.nextNonce = nonce + 1
+			for _, fillGap := ms.msgs[ms.nextNonce]; fillGap; _, fillGap = ms.msgs[ms.nextNonce] {
+				ms.nextNonce++
+			}
+		}
+		return
+	}
+
+	ms.requiredFunds.Sub(ms.requiredFunds, m.Message.RequiredFunds().Int)
+	//ms.requiredFunds.Sub(ms.requiredFunds, m.Message.Value.Int)
+	delete(ms.msgs, nonce)
+
+	// adjust next nonce
+	if applied {
+		// we removed a (known) message because it was applied in a tipset
+		// we can't possibly have filled a gap in this case
+		if nonce >= ms.nextNonce {
+			ms.nextNonce = nonce + 1
+		}
+		return
+	}
+
+	// we removed a message because it was pruned
+	// we have to adjust the nonce if it creates a gap or rewinds state
+	if nonce < ms.nextNonce {
+		ms.nextNonce = nonce
 	}
 }
 
@@ -204,9 +262,7 @@ func New(api Provider, ds dtypes.MetadataDS, netName dtypes.NetworkName) (*Messa
 
 	cfg, err := loadConfig(ds)
 	if err != nil {
-		if err != nil {
-			return nil, xerrors.Errorf("error loading mpool config: %w", err)
-		}
+		return nil, xerrors.Errorf("error loading mpool config: %w", err)
 	}
 
 	mp := &MessagePool{
@@ -478,6 +534,40 @@ func (mp *MessagePool) addTs(m *types.SignedMessage, curTs *types.TipSet) error
 	return mp.addLocked(m, true)
 }
 
+func (mp *MessagePool) addLoaded(m *types.SignedMessage) error {
+	err := mp.checkMessage(m)
+	if err != nil {
+		return err
+	}
+
+	mp.curTsLk.Lock()
+	defer mp.curTsLk.Unlock()
+
+	curTs := mp.curTs
+
+	snonce, err := mp.getStateNonce(m.Message.From, curTs)
+	if err != nil {
+		return xerrors.Errorf("failed to look up actor state nonce: %s: %w", err, ErrSoftValidationFailure)
+	}
+
+	if snonce > m.Message.Nonce {
+		return xerrors.Errorf("minimum expected nonce is %d: %w", snonce, ErrNonceTooLow)
+	}
+
+	mp.lk.Lock()
+	defer mp.lk.Unlock()
+
+	if err := mp.verifyMsgBeforeAdd(m, curTs.Height()); err != nil {
+		return err
+	}
+
+	if err := mp.checkBalance(m, curTs); err != nil {
+		return err
+	}
+
+	return mp.addLocked(m, false)
+}
+
 func (mp *MessagePool) addSkipChecks(m *types.SignedMessage) error {
 	mp.lk.Lock()
 	defer mp.lk.Unlock()
@@ -485,7 +575,7 @@ func (mp *MessagePool) addSkipChecks(m *types.SignedMessage) error {
 	return mp.addLocked(m, false)
 }
 
-func (mp *MessagePool) addLocked(m *types.SignedMessage, limit bool) error {
+func (mp *MessagePool) addLocked(m *types.SignedMessage, strict bool) error {
 	log.Debugf("mpooladd: %s %d", m.Message.From, m.Message.Nonce)
 	if m.Signature.Type == crypto.SigTypeBLS {
 		mp.blsSigCache.Add(m.Cid(), m.Signature)
@@ -503,11 +593,16 @@ func (mp *MessagePool) addLocked(m *types.SignedMessage, limit bool) error {
 
 	mset, ok := mp.pending[m.Message.From]
 	if !ok {
-		mset = newMsgSet()
+		nonce, err := mp.getStateNonce(m.Message.From, mp.curTs)
+		if err != nil {
+			return xerrors.Errorf("failed to get initial actor nonce: %w", err)
+		}
+
+		mset = newMsgSet(nonce)
 		mp.pending[m.Message.From] = mset
 	}
 
-	incr, err := mset.add(m, mp, limit)
+	incr, err := mset.add(m, mp, strict)
 	if err != nil {
 		log.Info(err)
 		return err
@@ -666,14 +761,14 @@ func (mp *MessagePool) PushWithNonce(ctx context.Context, addr address.Address,
 	return msg, mp.api.PubSubPublish(build.MessagesTopic(mp.netName), msgb)
 }
 
-func (mp *MessagePool) Remove(from address.Address, nonce uint64) {
+func (mp *MessagePool) Remove(from address.Address, nonce uint64, applied bool) {
 	mp.lk.Lock()
 	defer mp.lk.Unlock()
 
-	mp.remove(from, nonce)
+	mp.remove(from, nonce, applied)
 }
 
-func (mp *MessagePool) remove(from address.Address, nonce uint64) {
+func (mp *MessagePool) remove(from address.Address, nonce uint64, applied bool) {
 	mset, ok := mp.pending[from]
 	if !ok {
 		return
@@ -690,22 +785,10 @@ func (mp *MessagePool) remove(from address.Address, nonce uint64) {
 
 	// NB: This deletes any message with the given nonce. This makes sense
 	// as two messages with the same sender cannot have the same nonce
-	mset.rm(nonce)
+	mset.rm(nonce, applied)
 
 	if len(mset.msgs) == 0 {
 		delete(mp.pending, from)
-	} else {
-		var max uint64
-		for nonce := range mset.msgs {
-			if max < nonce {
-				max = nonce
-			}
-		}
-		if max < nonce {
-			max = nonce // we could have not seen the removed message before
-		}
-
-		mset.nextNonce = max + 1
 	}
 }
 
@@ -773,7 +856,7 @@ func (mp *MessagePool) HeadChange(revert []*types.TipSet, apply []*types.TipSet)
 	rm := func(from address.Address, nonce uint64) {
 		s, ok := rmsgs[from]
 		if !ok {
-			mp.Remove(from, nonce)
+			mp.Remove(from, nonce, true)
 			return
 		}
 
@@ -782,7 +865,7 @@ func (mp *MessagePool) HeadChange(revert []*types.TipSet, apply []*types.TipSet)
 			return
 		}
 
-		mp.Remove(from, nonce)
+		mp.Remove(from, nonce, true)
 	}
 
 	maybeRepub := func(cid cid.Cid) {
@@ -1084,7 +1167,7 @@ func (mp *MessagePool) loadLocal() error {
 			return xerrors.Errorf("unmarshaling local message: %w", err)
 		}
 
-		if err := mp.Add(&sm); err != nil {
+		if err := mp.addLoaded(&sm); err != nil {
 			if xerrors.Is(err, ErrNonceTooLow) {
 				continue // todo: drop the message from local cache (if above certain confidence threshold)
 			}

chain/messagepool/messagepool_test.go

@@ -352,6 +352,12 @@ func TestRevertMessages(t *testing.T) {
 }
 
 func TestPruningSimple(t *testing.T) {
+	oldMaxNonceGap := MaxNonceGap
+	MaxNonceGap = 1000
+	defer func() {
+		MaxNonceGap = oldMaxNonceGap
+	}()
+
 	tma := newTestMpoolAPI()
 
 	w, err := wallet.NewWallet(wallet.NewMemKeyStore())

chain/messagepool/pruning.go

@@ -98,7 +98,7 @@ keepLoop:
 	// and remove all messages that are still in pruneMsgs after processing the chains
 	log.Infof("Pruning %d messages", len(pruneMsgs))
 	for _, m := range pruneMsgs {
-		mp.remove(m.Message.From, m.Message.Nonce)
+		mp.remove(m.Message.From, m.Message.Nonce, false)
 	}
 
 	return nil

chain/messagepool/repub.go

@@ -3,6 +3,7 @@ package messagepool
 import (
 	"context"
 	"sort"
+	"time"
 
 	"golang.org/x/xerrors"
 
@@ -15,6 +16,8 @@ import (
 
 const repubMsgLimit = 30
 
+var RepublishBatchDelay = 200 * time.Millisecond
+
 func (mp *MessagePool) republishPendingMessages() error {
 	mp.curTsLk.Lock()
 	ts := mp.curTs
@@ -108,7 +111,7 @@ func (mp *MessagePool) republishPendingMessages() error {
 
 		// we can't fit the current chain but there is gas to spare
 		// trim it and push it down
-		chain.Trim(gasLimit, mp, baseFee, ts, false)
+		chain.Trim(gasLimit, mp, baseFee, ts)
 		for j := i; j < len(chains)-1; j++ {
 			if chains[j].Before(chains[j+1]) {
 				break
@@ -131,6 +134,12 @@ func (mp *MessagePool) republishPendingMessages() error {
 		}
 
 		count++
+
+		if count < len(msgs) {
+			// this delay is here to encourage the pubsub subsystem to process the messages serially
+			// and avoid creating nonce gaps because of concurrent validation.
+			time.Sleep(RepublishBatchDelay)
+		}
 	}
 
 	// track most recently republished messages

chain/messagepool/repub_test.go

@@ -12,6 +12,12 @@ import (
 )
 
 func TestRepubMessages(t *testing.T) {
+	oldRepublishBatchDelay := RepublishBatchDelay
+	RepublishBatchDelay = time.Microsecond
+	defer func() {
+		RepublishBatchDelay = oldRepublishBatchDelay
+	}()
+
 	tma := newTestMpoolAPI()
 	ds := datastore.NewMapDatastore()
 

chain/messagepool/selection.go

@@ -217,7 +217,7 @@ tailLoop:
 	for gasLimit >= minGas && last < len(chains) {
 		// trim if necessary
 		if chains[last].gasLimit > gasLimit {
-			chains[last].Trim(gasLimit, mp, baseFee, ts, false)
+			chains[last].Trim(gasLimit, mp, baseFee, ts)
 		}
 
 		// push down if it hasn't been invalidated
@@ -284,7 +284,7 @@ tailLoop:
 			}
 
 			// dependencies fit, just trim it
-			chain.Trim(gasLimit-depGasLimit, mp, baseFee, ts, false)
+			chain.Trim(gasLimit-depGasLimit, mp, baseFee, ts)
 			last += i
 			continue tailLoop
 		}
@@ -389,7 +389,7 @@ func (mp *MessagePool) selectMessagesGreedy(curTs, ts *types.TipSet) ([]*types.S
 tailLoop:
 	for gasLimit >= minGas && last < len(chains) {
 		// trim
-		chains[last].Trim(gasLimit, mp, baseFee, ts, false)
+		chains[last].Trim(gasLimit, mp, baseFee, ts)
 
 		// push down if it hasn't been invalidated
 		if chains[last].valid {
@@ -462,15 +462,27 @@ func (mp *MessagePool) selectPriorityMessages(pending map[address.Address]map[ui
 		}
 	}
 
+	if len(chains) == 0 {
+		return nil, gasLimit
+	}
+
 	// 2. Sort the chains
 	sort.Slice(chains, func(i, j int) bool {
 		return chains[i].Before(chains[j])
 	})
 
-	// 3. Merge chains until the block limit; we are willing to include negative performing chains
-	//    as these are messages from our own miners
+	if len(chains) != 0 && chains[0].gasPerf < 0 {
+		log.Warnw("all priority messages in mpool have negative gas performance", "bestGasPerf", chains[0].gasPerf)
+		return nil, gasLimit
+	}
+
+	// 3. Merge chains until the block limit, as long as they have non-negative gas performance
 	last := len(chains)
 	for i, chain := range chains {
+		if chain.gasPerf < 0 {
+			break
+		}
+
 		if chain.gasLimit <= gasLimit {
 			gasLimit -= chain.gasLimit
 			result = append(result, chain.msgs...)
@@ -484,8 +496,8 @@ func (mp *MessagePool) selectPriorityMessages(pending map[address.Address]map[ui
 
 tailLoop:
 	for gasLimit >= minGas && last < len(chains) {
-		// trim, without discarding negative performing messages
-		chains[last].Trim(gasLimit, mp, baseFee, ts, true)
+		// trim, discarding negative performing messages
+		chains[last].Trim(gasLimit, mp, baseFee, ts)
 
 		// push down if it hasn't been invalidated
 		if chains[last].valid {
@@ -503,6 +515,12 @@ tailLoop:
 			if !chain.valid {
 				continue
 			}
+
+			// if gasPerf < 0 we have no more profitable chains
+			if chain.gasPerf < 0 {
+				break tailLoop
+			}
+
 			// does it fit in the bock?
 			if chain.gasLimit <= gasLimit {
 				gasLimit -= chain.gasLimit
@@ -515,9 +533,9 @@ tailLoop:
 			continue tailLoop
 		}
 
-		// the merge loop ended after processing all the chains and we probably still have gas to spare
-		// -- mark the end.
-		last = len(chains)
+		// the merge loop ended after processing all the chains and we probably still have gas to spare;
+		// end the loop
+		break
 	}
 
 	return result, gasLimit
@@ -567,16 +585,18 @@ func (mp *MessagePool) getPendingMessages(curTs, ts *types.TipSet) (map[address.
 	return result, nil
 }
 
-func (mp *MessagePool) getGasReward(msg *types.SignedMessage, baseFee types.BigInt, ts *types.TipSet) *big.Int {
+func (*MessagePool) getGasReward(msg *types.SignedMessage, baseFee types.BigInt) *big.Int {
 	maxPremium := types.BigSub(msg.Message.GasFeeCap, baseFee)
-	if types.BigCmp(maxPremium, msg.Message.GasPremium) < 0 {
+
+	if types.BigCmp(maxPremium, msg.Message.GasPremium) > 0 {
 		maxPremium = msg.Message.GasPremium
 	}
+
 	gasReward := abig.Mul(maxPremium, types.NewInt(uint64(msg.Message.GasLimit)))
 	return gasReward.Int
 }
 
-func (mp *MessagePool) getGasPerf(gasReward *big.Int, gasLimit int64) float64 {
+func (*MessagePool) getGasPerf(gasReward *big.Int, gasLimit int64) float64 {
 	// gasPerf = gasReward * build.BlockGasLimit / gasLimit
 	a := new(big.Rat).SetInt(new(big.Int).Mul(gasReward, bigBlockGasLimit))
 	b := big.NewRat(1, gasLimit)
@@ -654,7 +674,7 @@ func (mp *MessagePool) createMessageChains(actor address.Address, mset map[uint6
 			balance = new(big.Int).Sub(balance, value)
 		}
 
-		gasReward := mp.getGasReward(m, baseFee, ts)
+		gasReward := mp.getGasReward(m, baseFee)
 		rewards = append(rewards, gasReward)
 	}
 
@@ -755,10 +775,10 @@ func (mc *msgChain) Before(other *msgChain) bool {
 		(mc.gasPerf == other.gasPerf && mc.gasReward.Cmp(other.gasReward) > 0)
 }
 
-func (mc *msgChain) Trim(gasLimit int64, mp *MessagePool, baseFee types.BigInt, ts *types.TipSet, priority bool) {
+func (mc *msgChain) Trim(gasLimit int64, mp *MessagePool, baseFee types.BigInt, ts *types.TipSet) {
 	i := len(mc.msgs) - 1
-	for i >= 0 && (mc.gasLimit > gasLimit || (!priority && mc.gasPerf < 0)) {
-		gasReward := mp.getGasReward(mc.msgs[i], baseFee, ts)
+	for i >= 0 && (mc.gasLimit > gasLimit || mc.gasPerf < 0) {
+		gasReward := mp.getGasReward(mc.msgs[i], baseFee)
 		mc.gasReward = new(big.Int).Sub(mc.gasReward, gasReward)
 		mc.gasLimit -= mc.msgs[i].Message.GasLimit
 		if mc.gasLimit > 0 {

chain/messagepool/selection_test.go

@@ -2,6 +2,7 @@ package messagepool
 
 import (
 	"context"
+	"fmt"
 	"math"
 	"math/big"
 	"math/rand"
@@ -369,6 +370,12 @@ func TestMessageChainSkipping(t *testing.T) {
 }
 
 func TestBasicMessageSelection(t *testing.T) {
+	oldMaxNonceGap := MaxNonceGap
+	MaxNonceGap = 1000
+	defer func() {
+		MaxNonceGap = oldMaxNonceGap
+	}()
+
 	mp, tma := makeTestMpool()
 
 	// the actors
@@ -1055,17 +1062,17 @@ func testCompetitiveMessageSelection(t *testing.T, rng *rand.Rand, getPremium fu
 
 		greedyReward := big.NewInt(0)
 		for _, m := range greedyMsgs {
-			greedyReward.Add(greedyReward, mp.getGasReward(m, baseFee, ts))
+			greedyReward.Add(greedyReward, mp.getGasReward(m, baseFee))
 		}
 
 		optReward := big.NewInt(0)
 		for _, m := range optMsgs {
-			optReward.Add(optReward, mp.getGasReward(m, baseFee, ts))
+			optReward.Add(optReward, mp.getGasReward(m, baseFee))
 		}
 
 		bestTqReward := big.NewInt(0)
 		for _, m := range bestMsgs {
-			bestTqReward.Add(bestTqReward, mp.getGasReward(m, baseFee, ts))
+			bestTqReward.Add(bestTqReward, mp.getGasReward(m, baseFee))
 		}
 
 		totalBestTQReward += float64(bestTqReward.Uint64())
@@ -1146,3 +1153,35 @@ func TestCompetitiveMessageSelectionZipf(t *testing.T) {
 	t.Logf("Average reward boost across all seeds: %f", rewardBoost)
 	t.Logf("Average reward of best ticket across all seeds: %f", tqReward)
 }
+
+func TestGasReward(t *testing.T) {
+	tests := []struct {
+		Premium   uint64
+		FeeCap    uint64
+		BaseFee   uint64
+		GasReward int64
+	}{
+		{Premium: 100, FeeCap: 200, BaseFee: 100, GasReward: 100},
+		{Premium: 100, FeeCap: 200, BaseFee: 210, GasReward: -10},
+		{Premium: 200, FeeCap: 250, BaseFee: 210, GasReward: 40},
+		{Premium: 200, FeeCap: 250, BaseFee: 2000, GasReward: -1750},
+	}
+
+	mp := new(MessagePool)
+	for _, test := range tests {
+		test := test
+		t.Run(fmt.Sprintf("%v", test), func(t *testing.T) {
+			msg := &types.SignedMessage{
+				Message: types.Message{
+					GasLimit:   10,
+					GasFeeCap:  types.NewInt(test.FeeCap),
+					GasPremium: types.NewInt(test.Premium),
+				},
+			}
+			rew := mp.getGasReward(msg, types.NewInt(test.BaseFee))
+			if rew.Cmp(big.NewInt(test.GasReward*10)) != 0 {
+				t.Errorf("bad reward: expected %d, got %s", test.GasReward*10, rew)
+			}
+		})
+	}
+}

chain/stmgr/call.go

@@ -95,7 +95,7 @@ func (sm *StateManager) Call(ctx context.Context, msg *types.Message, ts *types.
 
 	state := ts.ParentState()
 
-	r := store.NewChainRand(sm.cs, ts.Cids(), ts.Height())
+	r := store.NewChainRand(sm.cs, ts.Cids())
 
 	return sm.CallRaw(ctx, msg, state, r, ts.Height())
 }
@@ -113,7 +113,7 @@ func (sm *StateManager) CallWithGas(ctx context.Context, msg *types.Message, pri
 		return nil, xerrors.Errorf("computing tipset state: %w", err)
 	}
 
-	r := store.NewChainRand(sm.cs, ts.Cids(), ts.Height())
+	r := store.NewChainRand(sm.cs, ts.Cids())
 
 	if span.IsRecordingEvents() {
 		span.AddAttributes(

chain/stmgr/stmgr.go

@@ -338,7 +338,7 @@ func (sm *StateManager) computeTipSetState(ctx context.Context, ts *types.TipSet
 		cids[i] = v.Cid()
 	}
 
-	r := store.NewChainRand(sm.cs, cids, blks[0].Height)
+	r := store.NewChainRand(sm.cs, cids)
 
 	blkmsgs, err := sm.cs.BlockMsgsForTipset(ts)
 	if err != nil {

chain/stmgr/utils.go

@@ -432,7 +432,7 @@ func ComputeState(ctx context.Context, sm *StateManager, height abi.ChainEpoch,
 		return cid.Undef, nil, err
 	}
 
-	r := store.NewChainRand(sm.cs, ts.Cids(), height)
+	r := store.NewChainRand(sm.cs, ts.Cids())
 	vmopt := &vm.VMOpts{
 		StateBase:      base,
 		Epoch:          height,

chain/store/store.go

@@ -49,6 +49,7 @@ var chainHeadKey = dstore.NewKey("head")
 var blockValidationCacheKeyPrefix = dstore.NewKey("blockValidation")
 
 var DefaultTipSetCacheSize = 8192
+var DefaultMsgMetaCacheSize = 2048
 
 func init() {
 	if s := os.Getenv("LOTUS_CHAIN_TIPSET_CACHE"); s != "" {
@@ -58,6 +59,14 @@ func init() {
 		}
 		DefaultTipSetCacheSize = tscs
 	}
+
+	if s := os.Getenv("LOTUS_CHAIN_MSGMETA_CACHE"); s != "" {
+		mmcs, err := strconv.Atoi(s)
+		if err != nil {
+			log.Errorf("failed to parse 'LOTUS_CHAIN_MSGMETA_CACHE' env var: %s", err)
+		}
+		DefaultMsgMetaCacheSize = mmcs
+	}
 }
 
 // ReorgNotifee represents a callback that gets called upon reorgs.
@@ -97,7 +106,7 @@ type ChainStore struct {
 }
 
 func NewChainStore(bs bstore.Blockstore, ds dstore.Batching, vmcalls vm.SyscallBuilder) *ChainStore {
-	c, _ := lru.NewARC(2048)
+	c, _ := lru.NewARC(DefaultMsgMetaCacheSize)
 	tsc, _ := lru.NewARC(DefaultTipSetCacheSize)
 	cs := &ChainStore{
 		bs:       bs,
@@ -834,7 +843,7 @@ type mmCids struct {
 	secpk []cid.Cid
 }
 
-func (cs *ChainStore) readMsgMetaCids(mmc cid.Cid) ([]cid.Cid, []cid.Cid, error) {
+func (cs *ChainStore) ReadMsgMetaCids(mmc cid.Cid) ([]cid.Cid, []cid.Cid, error) {
 	o, ok := cs.mmCache.Get(mmc)
 	if ok {
 		mmcids := o.(*mmCids)
@@ -890,7 +899,7 @@ func (cs *ChainStore) GetPath(ctx context.Context, from types.TipSetKey, to type
 }
 
 func (cs *ChainStore) MessagesForBlock(b *types.BlockHeader) ([]*types.Message, []*types.SignedMessage, error) {
-	blscids, secpkcids, err := cs.readMsgMetaCids(b.Messages)
+	blscids, secpkcids, err := cs.ReadMsgMetaCids(b.Messages)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -1114,7 +1123,7 @@ func (cs *ChainStore) GetTipsetByHeight(ctx context.Context, h abi.ChainEpoch, t
 	return cs.LoadTipSet(lbts.Parents())
 }
 
-func recurseLinks(bs bstore.Blockstore, root cid.Cid, in []cid.Cid) ([]cid.Cid, error) {
+func recurseLinks(bs bstore.Blockstore, walked *cid.Set, root cid.Cid, in []cid.Cid) ([]cid.Cid, error) {
 	if root.Prefix().Codec != cid.DagCBOR {
 		return in, nil
 	}
@@ -1131,9 +1140,14 @@ func recurseLinks(bs bstore.Blockstore, root cid.Cid, in []cid.Cid) ([]cid.Cid,
 			return
 		}
 
+		// traversed this already...
+		if !walked.Visit(c) {
+			return
+		}
+
 		in = append(in, c)
 		var err error
-		in, err = recurseLinks(bs, c, in)
+		in, err = recurseLinks(bs, walked, c, in)
 		if err != nil {
 			rerr = err
 		}
@@ -1145,12 +1159,13 @@ func recurseLinks(bs bstore.Blockstore, root cid.Cid, in []cid.Cid) ([]cid.Cid,
 	return in, rerr
 }
 
-func (cs *ChainStore) Export(ctx context.Context, ts *types.TipSet, w io.Writer) error {
+func (cs *ChainStore) Export(ctx context.Context, ts *types.TipSet, inclRecentRoots abi.ChainEpoch, w io.Writer) error {
 	if ts == nil {
 		ts = cs.GetHeaviestTipSet()
 	}
 
 	seen := cid.NewSet()
+	walked := cid.NewSet()
 
 	h := &car.CarHeader{
 		Roots:   ts.Cids(),
@@ -1182,7 +1197,7 @@ func (cs *ChainStore) Export(ctx context.Context, ts *types.TipSet, w io.Writer)
 			return xerrors.Errorf("unmarshaling block header (cid=%s): %w", blk, err)
 		}
 
-		cids, err := recurseLinks(cs.bs, b.Messages, []cid.Cid{b.Messages})
+		cids, err := recurseLinks(cs.bs, walked, b.Messages, []cid.Cid{b.Messages})
 		if err != nil {
 			return xerrors.Errorf("recursing messages failed: %w", err)
 		}
@@ -1198,8 +1213,8 @@ func (cs *ChainStore) Export(ctx context.Context, ts *types.TipSet, w io.Writer)
 
 		out := cids
 
-		if b.Height == 0 {
-			cids, err := recurseLinks(cs.bs, b.ParentStateRoot, []cid.Cid{b.ParentStateRoot})
+		if b.Height == 0 || b.Height > ts.Height()-inclRecentRoots {
+			cids, err := recurseLinks(cs.bs, walked, b.ParentStateRoot, []cid.Cid{b.ParentStateRoot})
 			if err != nil {
 				return xerrors.Errorf("recursing genesis state failed: %w", err)
 			}
@@ -1282,14 +1297,12 @@ func (cs *ChainStore) GetLatestBeaconEntry(ts *types.TipSet) (*types.BeaconEntry
 type chainRand struct {
 	cs   *ChainStore
 	blks []cid.Cid
-	bh   abi.ChainEpoch
 }
 
-func NewChainRand(cs *ChainStore, blks []cid.Cid, bheight abi.ChainEpoch) vm.Rand {
+func NewChainRand(cs *ChainStore, blks []cid.Cid) vm.Rand {
 	return &chainRand{
 		cs:   cs,
 		blks: blks,
-		bh:   bheight,
 	}
 }
 

chain/store/store_test.go

@@ -96,7 +96,7 @@ func TestChainExportImport(t *testing.T) {
 	}
 
 	buf := new(bytes.Buffer)
-	if err := cg.ChainStore().Export(context.TODO(), last, buf); err != nil {
+	if err := cg.ChainStore().Export(context.TODO(), last, 0, buf); err != nil {
 		t.Fatal(err)
 	}
 

chain/store/weight.go

@@ -41,7 +41,7 @@ func (cs *ChainStore) Weight(ctx context.Context, ts *types.TipSet) (types.BigIn
 
 		var st power.State
 		if err := cst.Get(ctx, act.Head, &st); err != nil {
-			return types.NewInt(0), xerrors.Errorf("get power actor head: %w", err)
+			return types.NewInt(0), xerrors.Errorf("get power actor head (%s, height=%d): %w", act.Head, ts.Height(), err)
 		}
 		tpow = st.TotalQualityAdjPower // TODO: REVIEW: Is this correct?
 	}

chain/sub/incoming.go

@@ -555,6 +555,8 @@ func (mv *MessageValidator) Validate(ctx context.Context, pid peer.ID, msg *pubs
 			fallthrough
 		case xerrors.Is(err, messagepool.ErrTooManyPendingMessages):
 			fallthrough
+		case xerrors.Is(err, messagepool.ErrNonceGap):
+			fallthrough
 		case xerrors.Is(err, messagepool.ErrNonceTooLow):
 			return pubsub.ValidationIgnore
 		default:

cli/chain.go

@@ -859,6 +859,10 @@ var chainExportCmd = &cli.Command{
 		&cli.StringFlag{
 			Name: "tipset",
 		},
+		&cli.Int64Flag{
+			Name:  "recent-stateroots",
+			Usage: "specify the number of recent state roots to include in the export",
+		},
 	},
 	Action: func(cctx *cli.Context) error {
 		api, closer, err := GetFullNodeAPI(cctx)
@@ -872,6 +876,11 @@ var chainExportCmd = &cli.Command{
 			return fmt.Errorf("must specify filename to export chain to")
 		}
 
+		rsrs := abi.ChainEpoch(cctx.Int64("recent-stateroots"))
+		if cctx.IsSet("recent-stateroots") && rsrs < build.Finality {
+			return fmt.Errorf("\"recent-stateroots\" has to be greater than %d", build.Finality)
+		}
+
 		fi, err := os.Create(cctx.Args().First())
 		if err != nil {
 			return err
@@ -888,7 +897,7 @@ var chainExportCmd = &cli.Command{
 			return err
 		}
 
-		stream, err := api.ChainExport(ctx, ts.Key())
+		stream, err := api.ChainExport(ctx, rsrs, ts.Key())
 		if err != nil {
 			return err
 		}

cli/cmd.go

@@ -75,6 +75,8 @@ func flagForAPI(t repo.RepoType) string {
 		return "api"
 	case repo.StorageMiner:
 		return "miner-api"
+	case repo.Worker:
+		return "worker-api"
 	default:
 		panic(fmt.Sprintf("Unknown repo type: %v", t))
 	}
@@ -86,6 +88,8 @@ func flagForRepo(t repo.RepoType) string {
 		return "repo"
 	case repo.StorageMiner:
 		return "miner-repo"
+	case repo.Worker:
+		return "worker-repo"
 	default:
 		panic(fmt.Sprintf("Unknown repo type: %v", t))
 	}
@@ -97,6 +101,8 @@ func envForRepo(t repo.RepoType) string {
 		return "FULLNODE_API_INFO"
 	case repo.StorageMiner:
 		return "MINER_API_INFO"
+	case repo.Worker:
+		return "WORKER_API_INFO"
 	default:
 		panic(fmt.Sprintf("Unknown repo type: %v", t))
 	}
@@ -109,6 +115,8 @@ func envForRepoDeprecation(t repo.RepoType) string {
 		return "FULLNODE_API_INFO"
 	case repo.StorageMiner:
 		return "STORAGE_API_INFO"
+	case repo.Worker:
+		return "WORKER_API_INFO"
 	default:
 		panic(fmt.Sprintf("Unknown repo type: %v", t))
 	}
@@ -234,6 +242,15 @@ func GetStorageMinerAPI(ctx *cli.Context, opts ...jsonrpc.Option) (api.StorageMi
 	return client.NewStorageMinerRPC(ctx.Context, addr, headers, opts...)
 }
 
+func GetWorkerAPI(ctx *cli.Context) (api.WorkerAPI, jsonrpc.ClientCloser, error) {
+	addr, headers, err := GetRawAPI(ctx, repo.Worker)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	return client.NewWorkerRPC(ctx.Context, addr, headers)
+}
+
 func DaemonContext(cctx *cli.Context) context.Context {
 	if mtCtx, ok := cctx.App.Metadata[metadataTraceContext]; ok {
 		return mtCtx.(context.Context)

cli/net.go

@@ -6,9 +6,12 @@ import (
 	"os"
 	"sort"
 	"strings"
+	"text/tabwriter"
 
 	"github.com/libp2p/go-libp2p-core/peer"
+	protocol "github.com/libp2p/go-libp2p-core/protocol"
 
+	"github.com/dustin/go-humanize"
 	"github.com/urfave/cli/v2"
 
 	"github.com/filecoin-project/lotus/lib/addrutil"
@@ -25,6 +28,7 @@ var netCmd = &cli.Command{
 		netFindPeer,
 		netScores,
 		NetReachability,
+		NetBandwidthCmd,
 	},
 }
 
@@ -228,3 +232,88 @@ var NetReachability = &cli.Command{
 		return nil
 	},
 }
+
+var NetBandwidthCmd = &cli.Command{
+	Name:  "bandwidth",
+	Usage: "Print bandwidth usage information",
+	Flags: []cli.Flag{
+		&cli.BoolFlag{
+			Name:  "by-peer",
+			Usage: "list bandwidth usage by peer",
+		},
+		&cli.BoolFlag{
+			Name:  "by-protocol",
+			Usage: "list bandwidth usage by protocol",
+		},
+	},
+	Action: func(cctx *cli.Context) error {
+		api, closer, err := GetAPI(cctx)
+		if err != nil {
+			return err
+		}
+		defer closer()
+
+		ctx := ReqContext(cctx)
+
+		bypeer := cctx.Bool("by-peer")
+		byproto := cctx.Bool("by-protocol")
+
+		tw := tabwriter.NewWriter(os.Stdout, 4, 4, 2, ' ', 0)
+
+		fmt.Fprintf(tw, "Segment\tTotalIn\tTotalOut\tRateIn\tRateOut\n")
+
+		if bypeer {
+			bw, err := api.NetBandwidthStatsByPeer(ctx)
+			if err != nil {
+				return err
+			}
+
+			var peers []string
+			for p := range bw {
+				peers = append(peers, p)
+			}
+
+			sort.Slice(peers, func(i, j int) bool {
+				return peers[i] < peers[j]
+			})
+
+			for _, p := range peers {
+				s := bw[p]
+				fmt.Fprintf(tw, "%s\t%s\t%s\t%s/s\t%s/s\n", p, humanize.Bytes(uint64(s.TotalIn)), humanize.Bytes(uint64(s.TotalOut)), humanize.Bytes(uint64(s.RateIn)), humanize.Bytes(uint64(s.RateOut)))
+			}
+		} else if byproto {
+			bw, err := api.NetBandwidthStatsByProtocol(ctx)
+			if err != nil {
+				return err
+			}
+
+			var protos []protocol.ID
+			for p := range bw {
+				protos = append(protos, p)
+			}
+
+			sort.Slice(protos, func(i, j int) bool {
+				return protos[i] < protos[j]
+			})
+
+			for _, p := range protos {
+				s := bw[p]
+				if p == "" {
+					p = "<unknown>"
+				}
+				fmt.Fprintf(tw, "%s\t%s\t%s\t%s/s\t%s/s\n", p, humanize.Bytes(uint64(s.TotalIn)), humanize.Bytes(uint64(s.TotalOut)), humanize.Bytes(uint64(s.RateIn)), humanize.Bytes(uint64(s.RateOut)))
+			}
+		} else {
+
+			s, err := api.NetBandwidthStats(ctx)
+			if err != nil {
+				return err
+			}
+
+			fmt.Fprintf(tw, "Total\t%s\t%s\t%s/s\t%s/s\n", humanize.Bytes(uint64(s.TotalIn)), humanize.Bytes(uint64(s.TotalOut)), humanize.Bytes(uint64(s.RateIn)), humanize.Bytes(uint64(s.RateOut)))
+		}
+
+		return tw.Flush()
+
+	},
+}

cli/sync.go

@@ -180,11 +180,13 @@ func SyncWait(ctx context.Context, napi api.FullNode) error {
 		ss := state.ActiveSyncs[working]
 
 		var target []cid.Cid
+		var theight abi.ChainEpoch
 		if ss.Target != nil {
 			target = ss.Target.Cids()
+			theight = ss.Target.Height()
 		}
 
-		fmt.Printf("\r\x1b[2KWorker %d: Target: %s\tState: %s\tHeight: %d", working, target, chain.SyncStageString(ss.Stage), ss.Height)
+		fmt.Printf("\r\x1b[2KWorker %d: Target Height: %d\tTarget: %s\tState: %s\tHeight: %d", working, theight, target, chain.SyncStageString(ss.Stage), ss.Height)
 
 		if time.Now().Unix()-int64(head.MinTimestamp()) < int64(build.BlockDelaySecs) {
 			fmt.Println("\nDone!")

cmd/lotus-chainwatch/processor/processor.go

@@ -369,7 +369,9 @@ where rnk <= $1
 			maxBlock = bh.Height
 		}
 	}
-	log.Infow("Gathered Blocks to Process", "start", minBlock, "end", maxBlock)
+	if minBlock <= maxBlock {
+		log.Infow("Gathered Blocks to Process", "start", minBlock, "end", maxBlock)
+	}
 	return out, rows.Close()
 }
 

cmd/lotus-chainwatch/syncer/blockssub.go

@@ -11,16 +11,17 @@ import (
 func (s *Syncer) subBlocks(ctx context.Context) {
 	sub, err := s.node.SyncIncomingBlocks(ctx)
 	if err != nil {
-		log.Error(err)
+		log.Errorf("opening incoming block channel: %+v", err)
 		return
 	}
 
+	log.Infow("Capturing incoming blocks")
 	for bh := range sub {
 		err := s.storeHeaders(map[cid.Cid]*types.BlockHeader{
 			bh.Cid(): bh,
 		}, false, time.Now())
 		if err != nil {
-			log.Errorf("%+v", err)
+			log.Errorf("storing incoming block header: %+v", err)
 		}
 	}
 }

cmd/lotus-chainwatch/syncer/sync.go

@@ -160,23 +160,37 @@ func (s *Syncer) Start(ctx context.Context) {
 		log.Fatal(err)
 	}
 
+	// capture all reported blocks
+	go s.subBlocks(ctx)
+
+	// we need to ensure that on a restart we don't reprocess the whole flarping chain
+	var sinceEpoch uint64
+	blkCID, height, err := s.mostRecentlySyncedBlockHeight()
+	if err != nil {
+		log.Fatalw("failed to find most recently synced block", "error", err)
+	} else {
+		if height > 0 {
+			log.Infow("Found starting point for syncing", "blockCID", blkCID.String(), "height", height)
+			sinceEpoch = uint64(height)
+		}
+	}
+
 	// continue to keep the block headers table up to date.
 	notifs, err := s.node.ChainNotify(ctx)
 	if err != nil {
 		log.Fatal(err)
 	}
 
-	// we need to ensure that on a restart we don't reprocess the whole flarping chain
-	blkCID, height, err := s.mostRecentlySyncedBlockHeight()
-	if err != nil {
-		log.Fatalw("failed to find most recently synced block", "error", err)
-	}
-	log.Infow("Found starting point for syncing", "blockCID", blkCID.String(), "height", height)
-	sinceEpoch := uint64(height)
 	go func() {
 		for notif := range notifs {
 			for _, change := range notif {
 				switch change.Type {
+				case store.HCCurrent:
+					// This case is important for capturing the initial state of a node
+					// which might be on a dead network with no new blocks being produced.
+					// It also allows a fresh Chainwatch instance to start walking the
+					// chain without waiting for a new block to come along.
+					fallthrough
 				case store.HCApply:
 					unsynced, err := s.unsyncedBlocks(ctx, change.Val, sinceEpoch)
 					if err != nil {

cmd/lotus-pcr/main.go

@@ -133,6 +133,18 @@ var runCmd = &cli.Command{
 			Usage:   "do not send any messages",
 			Value:   false,
 		},
+		&cli.BoolFlag{
+			Name:    "pre-commit",
+			EnvVars: []string{"LOTUS_PCR_PRE_COMMIT"},
+			Usage:   "process PreCommitSector messages",
+			Value:   true,
+		},
+		&cli.BoolFlag{
+			Name:    "prove-commit",
+			EnvVars: []string{"LOTUS_PCR_PROVE_COMMIT"},
+			Usage:   "process ProveCommitSector messages",
+			Value:   true,
+		},
 		&cli.IntFlag{
 			Name:    "head-delay",
 			EnvVars: []string{"LOTUS_PCR_HEAD_DELAY"},
@@ -180,12 +192,16 @@ var runCmd = &cli.Command{
 		percentExtra := cctx.Int("percent-extra")
 		maxMessageQueue := cctx.Int("max-message-queue")
 		dryRun := cctx.Bool("dry-run")
+		preCommitEnabled := cctx.Bool("pre-commit")
+		proveCommitEnabled := cctx.Bool("prove-commit")
 
 		rf := &refunder{
-			api:          api,
-			wallet:       from,
-			percentExtra: percentExtra,
-			dryRun:       dryRun,
+			api:                api,
+			wallet:             from,
+			percentExtra:       percentExtra,
+			dryRun:             dryRun,
+			preCommitEnabled:   preCommitEnabled,
+			proveCommitEnabled: proveCommitEnabled,
 		}
 
 		for tipset := range tipsetsCh {
@@ -281,10 +297,12 @@ type refunderNodeApi interface {
 }
 
 type refunder struct {
-	api          refunderNodeApi
-	wallet       address.Address
-	percentExtra int
-	dryRun       bool
+	api                refunderNodeApi
+	wallet             address.Address
+	percentExtra       int
+	dryRun             bool
+	preCommitEnabled   bool
+	proveCommitEnabled bool
 }
 
 func (r *refunder) ProcessTipset(ctx context.Context, tipset *types.TipSet) (*MinersRefund, error) {
@@ -331,7 +349,12 @@ func (r *refunder) ProcessTipset(ctx context.Context, tipset *types.TipSet) (*Mi
 
 		switch m.Method {
 		case builtin.MethodsMiner.ProveCommitSector:
+			if !r.proveCommitEnabled {
+				continue
+			}
+
 			messageMethod = "ProveCommitSector"
+
 			if recps[i].ExitCode != exitcode.Ok {
 				log.Debugw("skipping non-ok exitcode message", "method", messageMethod, "cid", msg.Cid, "miner", m.To, "exitcode", recps[i].ExitCode)
 				continue
@@ -369,6 +392,10 @@ func (r *refunder) ProcessTipset(ctx context.Context, tipset *types.TipSet) (*Mi
 
 			refundValue = collateral
 		case builtin.MethodsMiner.PreCommitSector:
+			if !r.preCommitEnabled {
+				continue
+			}
+
 			messageMethod = "PreCommitSector"
 
 			if recps[i].ExitCode != exitcode.Ok {

cmd/lotus-seal-worker/info.go

@@ -0,0 +1,71 @@
+package main
+
+import (
+	"fmt"
+
+	"github.com/urfave/cli/v2"
+	"golang.org/x/xerrors"
+
+	"github.com/filecoin-project/lotus/chain/types"
+	lcli "github.com/filecoin-project/lotus/cli"
+)
+
+var infoCmd = &cli.Command{
+	Name:  "info",
+	Usage: "Print worker info",
+	Action: func(cctx *cli.Context) error {
+		api, closer, err := lcli.GetWorkerAPI(cctx)
+		if err != nil {
+			return err
+		}
+		defer closer()
+
+		ctx := lcli.ReqContext(cctx)
+
+		ver, err := api.Version(ctx)
+		if err != nil {
+			return xerrors.Errorf("getting version: %w", err)
+		}
+
+		fmt.Println("Worker version: ", ver)
+		fmt.Print("CLI version: ")
+		cli.VersionPrinter(cctx)
+		fmt.Println()
+
+		info, err := api.Info(ctx)
+		if err != nil {
+			return xerrors.Errorf("getting info: %w", err)
+		}
+
+		fmt.Printf("Hostname: %s\n", info.Hostname)
+		fmt.Printf("CPUs: %d; GPUs: %v\n", info.Resources.CPUs, info.Resources.GPUs)
+		fmt.Printf("RAM: %s; Swap: %s\n", types.SizeStr(types.NewInt(info.Resources.MemPhysical)), types.SizeStr(types.NewInt(info.Resources.MemSwap)))
+		fmt.Printf("Reserved memory: %s\n", types.SizeStr(types.NewInt(info.Resources.MemReserved)))
+		fmt.Println()
+
+		paths, err := api.Paths(ctx)
+		if err != nil {
+			return xerrors.Errorf("getting path info: %w", err)
+		}
+
+		for _, path := range paths {
+			fmt.Printf("%s:\n", path.ID)
+			fmt.Printf("\tWeight: %d; Use: ", path.Weight)
+			if path.CanSeal || path.CanStore {
+				fmt.Printf("Weight: %d; Use: ", path.Weight)
+				if path.CanSeal {
+					fmt.Print("Seal ")
+				}
+				if path.CanStore {
+					fmt.Print("Store")
+				}
+				fmt.Println("")
+			} else {
+				fmt.Print("Use: ReadOnly")
+			}
+			fmt.Printf("\tLocal: %s\n", path.LocalPath)
+		}
+
+		return nil
+	},
+}

cmd/lotus-seal-worker/main.go

@@ -16,6 +16,7 @@ import (
 	"github.com/google/uuid"
 	"github.com/gorilla/mux"
 	logging "github.com/ipfs/go-log/v2"
+	manet "github.com/multiformats/go-multiaddr/net"
 	"github.com/urfave/cli/v2"
 	"golang.org/x/xerrors"
 
@@ -46,10 +47,10 @@ const FlagWorkerRepoDeprecation = "workerrepo"
 func main() {
 	lotuslog.SetupLogLevels()
 
-	log.Info("Starting lotus worker")
-
 	local := []*cli.Command{
 		runCmd,
+		infoCmd,
+		storageCmd,
 	}
 
 	app := &cli.App{
@@ -153,6 +154,8 @@ var runCmd = &cli.Command{
 		return nil
 	},
 	Action: func(cctx *cli.Context) error {
+		log.Info("Starting lotus worker")
+
 		if !cctx.Bool("enable-gpu-proving") {
 			if err := os.Setenv("BELLMAN_NO_GPU", "true"); err != nil {
 				return xerrors.Errorf("could not set no-gpu env: %+v", err)
@@ -342,6 +345,8 @@ var runCmd = &cli.Command{
 				SealProof: spt,
 				TaskTypes: taskTypes,
 			}, remote, localStore, nodeApi),
+			localStore: localStore,
+			ls:         lr,
 		}
 
 		mux := mux.NewRouter()
@@ -383,6 +388,32 @@ var runCmd = &cli.Command{
 			return err
 		}
 
+		{
+			a, err := net.ResolveTCPAddr("tcp", address)
+			if err != nil {
+				return xerrors.Errorf("parsing address: %w", err)
+			}
+
+			ma, err := manet.FromNetAddr(a)
+			if err != nil {
+				return xerrors.Errorf("creating api multiaddress: %w", err)
+			}
+
+			if err := lr.SetAPIEndpoint(ma); err != nil {
+				return xerrors.Errorf("setting api endpoint: %w", err)
+			}
+
+			ainfo, err := lcli.GetAPIInfo(cctx, repo.StorageMiner)
+			if err != nil {
+				return xerrors.Errorf("could not get miner API info: %w", err)
+			}
+
+			// TODO: ideally this would be a token with some permissions dropped
+			if err := lr.SetAPIToken(ainfo.Token); err != nil {
+				return xerrors.Errorf("setting api token: %w", err)
+			}
+		}
+
 		log.Info("Waiting for tasks")
 
 		go func() {

cmd/lotus-seal-worker/rpc.go

@@ -3,19 +3,44 @@ package main
 import (
 	"context"
 
+	"github.com/mitchellh/go-homedir"
+	"golang.org/x/xerrors"
+
 	"github.com/filecoin-project/specs-storage/storage"
 
-	sectorstorage "github.com/filecoin-project/lotus/extern/sector-storage"
-
 	"github.com/filecoin-project/lotus/build"
+	sectorstorage "github.com/filecoin-project/lotus/extern/sector-storage"
+	"github.com/filecoin-project/lotus/extern/sector-storage/stores"
 )
 
 type worker struct {
 	*sectorstorage.LocalWorker
+
+	localStore *stores.Local
+	ls         stores.LocalStorage
 }
 
 func (w *worker) Version(context.Context) (build.Version, error) {
 	return build.APIVersion, nil
 }
 
+func (w *worker) StorageAddLocal(ctx context.Context, path string) error {
+	path, err := homedir.Expand(path)
+	if err != nil {
+		return xerrors.Errorf("expanding local path: %w", err)
+	}
+
+	if err := w.localStore.OpenPath(ctx, path); err != nil {
+		return xerrors.Errorf("opening local path: %w", err)
+	}
+
+	if err := w.ls.SetStorage(func(sc *stores.StorageConfig) {
+		sc.StoragePaths = append(sc.StoragePaths, stores.LocalPath{Path: path})
+	}); err != nil {
+		return xerrors.Errorf("get storage config: %w", err)
+	}
+
+	return nil
+}
+
 var _ storage.Sealer = &worker{}

cmd/lotus-seal-worker/storage.go

@@ -0,0 +1,105 @@
+package main
+
+import (
+	"encoding/json"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+
+	"github.com/google/uuid"
+	"github.com/mitchellh/go-homedir"
+	"github.com/urfave/cli/v2"
+	"golang.org/x/xerrors"
+
+	lcli "github.com/filecoin-project/lotus/cli"
+	"github.com/filecoin-project/lotus/extern/sector-storage/stores"
+)
+
+const metaFile = "sectorstore.json"
+
+var storageCmd = &cli.Command{
+	Name:  "storage",
+	Usage: "manage sector storage",
+	Subcommands: []*cli.Command{
+		storageAttachCmd,
+	},
+}
+
+var storageAttachCmd = &cli.Command{
+	Name:  "attach",
+	Usage: "attach local storage path",
+	Flags: []cli.Flag{
+		&cli.BoolFlag{
+			Name:  "init",
+			Usage: "initialize the path first",
+		},
+		&cli.Uint64Flag{
+			Name:  "weight",
+			Usage: "(for init) path weight",
+			Value: 10,
+		},
+		&cli.BoolFlag{
+			Name:  "seal",
+			Usage: "(for init) use path for sealing",
+		},
+		&cli.BoolFlag{
+			Name:  "store",
+			Usage: "(for init) use path for long-term storage",
+		},
+	},
+	Action: func(cctx *cli.Context) error {
+		nodeApi, closer, err := lcli.GetWorkerAPI(cctx)
+		if err != nil {
+			return err
+		}
+		defer closer()
+		ctx := lcli.ReqContext(cctx)
+
+		if !cctx.Args().Present() {
+			return xerrors.Errorf("must specify storage path to attach")
+		}
+
+		p, err := homedir.Expand(cctx.Args().First())
+		if err != nil {
+			return xerrors.Errorf("expanding path: %w", err)
+		}
+
+		if cctx.Bool("init") {
+			if err := os.MkdirAll(p, 0755); err != nil {
+				if !os.IsExist(err) {
+					return err
+				}
+			}
+
+			_, err := os.Stat(filepath.Join(p, metaFile))
+			if !os.IsNotExist(err) {
+				if err == nil {
+					return xerrors.Errorf("path is already initialized")
+				}
+				return err
+			}
+
+			cfg := &stores.LocalStorageMeta{
+				ID:       stores.ID(uuid.New().String()),
+				Weight:   cctx.Uint64("weight"),
+				CanSeal:  cctx.Bool("seal"),
+				CanStore: cctx.Bool("store"),
+			}
+
+			if !(cfg.CanStore || cfg.CanSeal) {
+				return xerrors.Errorf("must specify at least one of --store of --seal")
+			}
+
+			b, err := json.MarshalIndent(cfg, "", "  ")
+			if err != nil {
+				return xerrors.Errorf("marshaling storage config: %w", err)
+			}
+
+			if err := ioutil.WriteFile(filepath.Join(p, metaFile), b, 0644); err != nil {
+				return xerrors.Errorf("persisting storage metadata (%s): %w", filepath.Join(p, metaFile), err)
+			}
+		}
+
+		return nodeApi.StorageAddLocal(ctx, p)
+	},
+}

cmd/lotus/daemon.go

@@ -100,7 +100,11 @@ var DaemonCmd = &cli.Command{
 		},
 		&cli.StringFlag{
 			Name:  "import-chain",
-			Usage: "on first run, load chain from given file",
+			Usage: "on first run, load chain from given file and validate",
+		},
+		&cli.StringFlag{
+			Name:  "import-snapshot",
+			Usage: "import chain state from a given chain export file",
 		},
 		&cli.BoolFlag{
 			Name:  "halt-after-import",
@@ -191,13 +195,23 @@ var DaemonCmd = &cli.Command{
 		}
 
 		chainfile := cctx.String("import-chain")
-		if chainfile != "" {
+		snapshot := cctx.String("import-snapshot")
+		if chainfile != "" || snapshot != "" {
+			if chainfile != "" && snapshot != "" {
+				return fmt.Errorf("cannot specify both 'import-snapshot' and 'import-chain'")
+			}
+			var issnapshot bool
+			if chainfile == "" {
+				chainfile = snapshot
+				issnapshot = true
+			}
+
 			chainfile, err := homedir.Expand(chainfile)
 			if err != nil {
 				return err
 			}
 
-			if err := ImportChain(r, chainfile); err != nil {
+			if err := ImportChain(r, chainfile, issnapshot); err != nil {
 				return err
 			}
 			if cctx.Bool("halt-after-import") {
@@ -312,7 +326,7 @@ func importKey(ctx context.Context, api api.FullNode, f string) error {
 	return nil
 }
 
-func ImportChain(r repo.Repo, fname string) error {
+func ImportChain(r repo.Repo, fname string, snapshot bool) error {
 	fi, err := os.Open(fname)
 	if err != nil {
 		return err
@@ -357,9 +371,11 @@ func ImportChain(r repo.Repo, fname string) error {
 
 	stm := stmgr.NewStateManager(cst)
 
-	log.Infof("validating imported chain...")
-	if err := stm.ValidateChain(context.TODO(), ts); err != nil {
-		return xerrors.Errorf("chain validation failed: %w", err)
+	if !snapshot {
+		log.Infof("validating imported chain...")
+		if err := stm.ValidateChain(context.TODO(), ts); err != nil {
+			return xerrors.Errorf("chain validation failed: %w", err)
+		}
 	}
 
 	log.Info("accepting %s as new head", ts.Cids())

cmd/lotus/debug_advance.go

@@ -62,6 +62,9 @@ func init() {
 			}
 
 			mbi, err := api.MinerGetBaseInfo(ctx, addr, head.Height()+1, head.Key())
+			if err != nil {
+				return xerrors.Errorf("getting base info: %w", err)
+			}
 
 			ep := &types.ElectionProof{}
 			ep.WinCount = ep.ComputeWinCount(types.NewInt(1), types.NewInt(1))

conformance/driver.go

@@ -2,7 +2,10 @@ package conformance
 
 import (
 	"context"
+	"fmt"
 
+	"github.com/filecoin-project/lotus/chain/stmgr"
+	"github.com/filecoin-project/lotus/chain/store"
 	"github.com/filecoin-project/lotus/chain/types"
 	"github.com/filecoin-project/lotus/chain/vm"
 	"github.com/filecoin-project/lotus/extern/sector-storage/ffiwrapper"
@@ -14,7 +17,10 @@ import (
 	"github.com/filecoin-project/test-vectors/chaos"
 	"github.com/filecoin-project/test-vectors/schema"
 
+	"github.com/filecoin-project/go-address"
+
 	"github.com/ipfs/go-cid"
+	ds "github.com/ipfs/go-datastore"
 )
 
 var (
@@ -24,16 +30,92 @@ var (
 )
 
 type Driver struct {
-	ctx    context.Context
-	vector *schema.TestVector
+	ctx      context.Context
+	selector schema.Selector
 }
 
-func NewDriver(ctx context.Context, vector *schema.TestVector) *Driver {
-	return &Driver{ctx: ctx, vector: vector}
+func NewDriver(ctx context.Context, selector schema.Selector) *Driver {
+	return &Driver{ctx: ctx, selector: selector}
+}
+
+type ExecuteTipsetResult struct {
+	ReceiptsRoot  cid.Cid
+	PostStateRoot cid.Cid
+
+	// AppliedMessages stores the messages that were applied, in the order they
+	// were applied. It includes implicit messages (cron, rewards).
+	AppliedMessages []*types.Message
+	// AppliedResults stores the results of AppliedMessages, in the same order.
+	AppliedResults []*vm.ApplyRet
+}
+
+// ExecuteTipset executes the supplied tipset on top of the state represented
+// by the preroot CID.
+//
+// parentEpoch is the last epoch in which an actual tipset was processed. This
+// is used by Lotus for null block counting and cron firing.
+//
+// This method returns the the receipts root, the poststate root, and the VM
+// message results. The latter _include_ implicit messages, such as cron ticks
+// and reward withdrawal per miner.
+func (d *Driver) ExecuteTipset(bs blockstore.Blockstore, ds ds.Batching, preroot cid.Cid, parentEpoch abi.ChainEpoch, tipset *schema.Tipset) (*ExecuteTipsetResult, error) {
+	var (
+		syscalls = mkFakedSigSyscalls(vm.Syscalls(ffiwrapper.ProofVerifier))
+		vmRand   = new(testRand)
+
+		cs = store.NewChainStore(bs, ds, syscalls)
+		sm = stmgr.NewStateManager(cs)
+	)
+
+	blocks := make([]store.BlockMessages, 0, len(tipset.Blocks))
+	for _, b := range tipset.Blocks {
+		sb := store.BlockMessages{
+			Miner:    b.MinerAddr,
+			WinCount: b.WinCount,
+		}
+		for _, m := range b.Messages {
+			msg, err := types.DecodeMessage(m)
+			if err != nil {
+				return nil, err
+			}
+			switch msg.From.Protocol() {
+			case address.SECP256K1:
+				sb.SecpkMessages = append(sb.SecpkMessages, msg)
+			case address.BLS:
+				sb.BlsMessages = append(sb.BlsMessages, msg)
+			default:
+				return nil, fmt.Errorf("from account is not secpk nor bls: %s", msg.From)
+			}
+		}
+		blocks = append(blocks, sb)
+	}
+
+	var (
+		messages []*types.Message
+		results  []*vm.ApplyRet
+	)
+
+	postcid, receiptsroot, err := sm.ApplyBlocks(context.Background(), parentEpoch, preroot, blocks, tipset.Epoch, vmRand, func(_ cid.Cid, msg *types.Message, ret *vm.ApplyRet) error {
+		messages = append(messages, msg)
+		results = append(results, ret)
+		return nil
+	}, tipset.BaseFee)
+
+	if err != nil {
+		return nil, err
+	}
+
+	ret := &ExecuteTipsetResult{
+		ReceiptsRoot:    receiptsroot,
+		PostStateRoot:   postcid,
+		AppliedMessages: messages,
+		AppliedResults:  results,
+	}
+	return ret, nil
 }
 
 // ExecuteMessage executes a conformance test vector message in a temporary VM.
-func (d *Driver) ExecuteMessage(msg *types.Message, preroot cid.Cid, bs blockstore.Blockstore, epoch abi.ChainEpoch) (*vm.ApplyRet, cid.Cid, error) {
+func (d *Driver) ExecuteMessage(bs blockstore.Blockstore, preroot cid.Cid, epoch abi.ChainEpoch, msg *types.Message) (*vm.ApplyRet, cid.Cid, error) {
 	vmOpts := &vm.VMOpts{
 		StateBase:      preroot,
 		Epoch:          epoch,
@@ -52,10 +134,10 @@ func (d *Driver) ExecuteMessage(msg *types.Message, preroot cid.Cid, bs blocksto
 	invoker := vm.NewInvoker()
 
 	// add support for the puppet and chaos actors.
-	if puppetOn, ok := d.vector.Selector["puppet_actor"]; ok && puppetOn == "true" {
+	if puppetOn, ok := d.selector["puppet_actor"]; ok && puppetOn == "true" {
 		invoker.Register(puppet.PuppetActorCodeID, puppet.Actor{}, puppet.State{})
 	}
-	if chaosOn, ok := d.vector.Selector["chaos_actor"]; ok && chaosOn == "true" {
+	if chaosOn, ok := d.selector["chaos_actor"]; ok && chaosOn == "true" {
 		invoker.Register(chaos.ChaosActorCodeCID, chaos.Actor{}, chaos.State{})
 	}
 

conformance/runner_test.go

@@ -6,12 +6,17 @@ import (
 	"context"
 	"encoding/base64"
 	"encoding/json"
+	"fmt"
 	"io/ioutil"
 	"os"
 	"path/filepath"
+	"strconv"
 	"strings"
 	"testing"
 
+	"github.com/ipfs/go-cid"
+	ds "github.com/ipfs/go-datastore"
+
 	"github.com/filecoin-project/lotus/chain/types"
 	"github.com/filecoin-project/lotus/chain/vm"
 	"github.com/filecoin-project/lotus/lib/blockstore"
@@ -135,6 +140,8 @@ func TestConformance(t *testing.T) {
 			switch vector.Class {
 			case "message":
 				executeMessageVector(t, &vector)
+			case "tipset":
+				executeTipsetVector(t, &vector)
 			default:
 				t.Fatalf("test vector class not supported: %s", vector.Class)
 			}
@@ -150,24 +157,11 @@ func executeMessageVector(t *testing.T, vector *schema.TestVector) {
 		root  = vector.Pre.StateTree.RootCID
 	)
 
-	bs := blockstore.NewTemporary()
-
-	// Read the base64-encoded CAR from the vector, and inflate the gzip.
-	buf := bytes.NewReader(vector.CAR)
-	r, err := gzip.NewReader(buf)
-	if err != nil {
-		t.Fatalf("failed to inflate gzipped CAR: %s", err)
-	}
-	defer r.Close() // nolint
-
-	// Load the CAR embedded in the test vector into the Blockstore.
-	_, err = car.LoadCar(bs, r)
-	if err != nil {
-		t.Fatalf("failed to load state tree car from test vector: %s", err)
-	}
+	// Load the CAR into a new temporary Blockstore.
+	bs := loadCAR(t, vector.CAR)
 
 	// Create a new Driver.
-	driver := NewDriver(ctx, vector)
+	driver := NewDriver(ctx, vector.Selector)
 
 	// Apply every message.
 	for i, m := range vector.ApplyMessages {
@@ -183,52 +177,128 @@ func executeMessageVector(t *testing.T, vector *schema.TestVector) {
 
 		// Execute the message.
 		var ret *vm.ApplyRet
-		ret, root, err = driver.ExecuteMessage(msg, root, bs, epoch)
+		ret, root, err = driver.ExecuteMessage(bs, root, epoch, msg)
 		if err != nil {
 			t.Fatalf("fatal failure when executing message: %s", err)
 		}
 
 		// Assert that the receipt matches what the test vector expects.
-		receipt := vector.Post.Receipts[i]
-		if expected, actual := receipt.ExitCode, ret.ExitCode; expected != actual {
-			t.Errorf("exit code of msg %d did not match; expected: %s, got: %s", i, expected, actual)
-		}
-		if expected, actual := receipt.GasUsed, ret.GasUsed; expected != actual {
-			t.Errorf("gas used of msg %d did not match; expected: %d, got: %d", i, expected, actual)
-		}
-		if expected, actual := []byte(receipt.ReturnValue), ret.Return; !bytes.Equal(expected, actual) {
-			t.Errorf("return value of msg %d did not match; expected: %s, got: %s", i, base64.StdEncoding.EncodeToString(expected), base64.StdEncoding.EncodeToString(actual))
-		}
+		assertMsgResult(t, vector.Post.Receipts[i], ret, strconv.Itoa(i))
 	}
 
 	// Once all messages are applied, assert that the final state root matches
 	// the expected postcondition root.
 	if root != vector.Post.StateTree.RootCID {
-		color.NoColor = false // enable colouring.
-
-		t.Errorf("wrong post root cid; expected %v, but got %v", vector.Post.StateTree.RootCID, root)
-
-		var (
-			a  = color.New(color.FgMagenta, color.Bold).Sprint("(A) expected final state")
-			b  = color.New(color.FgYellow, color.Bold).Sprint("(B) actual final state")
-			c  = color.New(color.FgCyan, color.Bold).Sprint("(C) initial state")
-			d1 = color.New(color.FgGreen, color.Bold).Sprint("[Δ1]")
-			d2 = color.New(color.FgGreen, color.Bold).Sprint("[Δ2]")
-			d3 = color.New(color.FgGreen, color.Bold).Sprint("[Δ3]")
-		)
-
-		bold := color.New(color.Bold).SprintfFunc()
-
-		// run state diffs.
-		t.Log(bold("=== dumping 3-way diffs between %s, %s, %s ===", a, b, c))
-
-		t.Log(bold("--- %s left: %s; right: %s ---", d1, a, b))
-		t.Log(statediff.Diff(context.Background(), bs, vector.Post.StateTree.RootCID, root))
-
-		t.Log(bold("--- %s left: %s; right: %s ---", d2, c, b))
-		t.Log(statediff.Diff(context.Background(), bs, vector.Pre.StateTree.RootCID, root))
-
-		t.Log(bold("--- %s left: %s; right: %s ---", d3, c, a))
-		t.Log(statediff.Diff(context.Background(), bs, vector.Pre.StateTree.RootCID, vector.Post.StateTree.RootCID))
+		dumpThreeWayStateDiff(t, vector, bs, root)
 	}
 }
+
+// executeTipsetVector executes a tipset-class test vector.
+func executeTipsetVector(t *testing.T, vector *schema.TestVector) {
+	var (
+		ctx       = context.Background()
+		prevEpoch = vector.Pre.Epoch
+		root      = vector.Pre.StateTree.RootCID
+		tmpds     = ds.NewMapDatastore()
+	)
+
+	// Load the CAR into a new temporary Blockstore.
+	bs := loadCAR(t, vector.CAR)
+
+	// Create a new Driver.
+	driver := NewDriver(ctx, vector.Selector)
+
+	// Apply every tipset.
+	var receiptsIdx int
+	for i, ts := range vector.ApplyTipsets {
+		ts := ts // capture
+		ret, err := driver.ExecuteTipset(bs, tmpds, root, prevEpoch, &ts)
+		if err != nil {
+			t.Fatalf("failed to apply tipset %d message: %s", i, err)
+		}
+
+		for j, v := range ret.AppliedResults {
+			assertMsgResult(t, vector.Post.Receipts[receiptsIdx], v, fmt.Sprintf("%d of tipset %d", j, i))
+			receiptsIdx++
+		}
+
+		// Compare the receipts root.
+		if expected, actual := vector.Post.ReceiptsRoots[i], ret.ReceiptsRoot; expected != actual {
+			t.Errorf("post receipts root doesn't match; expected: %s, was: %s", expected, actual)
+		}
+
+		prevEpoch = ts.Epoch
+		root = ret.PostStateRoot
+	}
+
+	// Once all messages are applied, assert that the final state root matches
+	// the expected postcondition root.
+	if root != vector.Post.StateTree.RootCID {
+		dumpThreeWayStateDiff(t, vector, bs, root)
+	}
+}
+
+// assertMsgResult compares a message result. It takes the expected receipt
+// encoded in the vector, the actual receipt returned by Lotus, and a message
+// label to log in the assertion failure message to facilitate debugging.
+func assertMsgResult(t *testing.T, expected *schema.Receipt, actual *vm.ApplyRet, label string) {
+	t.Helper()
+
+	if expected, actual := expected.ExitCode, actual.ExitCode; expected != actual {
+		t.Errorf("exit code of msg %s did not match; expected: %s, got: %s", label, expected, actual)
+	}
+	if expected, actual := expected.GasUsed, actual.GasUsed; expected != actual {
+		t.Errorf("gas used of msg %s did not match; expected: %d, got: %d", label, expected, actual)
+	}
+	if expected, actual := []byte(expected.ReturnValue), actual.Return; !bytes.Equal(expected, actual) {
+		t.Errorf("return value of msg %s did not match; expected: %s, got: %s", label, base64.StdEncoding.EncodeToString(expected), base64.StdEncoding.EncodeToString(actual))
+	}
+}
+
+func dumpThreeWayStateDiff(t *testing.T, vector *schema.TestVector, bs blockstore.Blockstore, actual cid.Cid) {
+	color.NoColor = false // enable colouring.
+
+	t.Errorf("wrong post root cid; expected %v, but got %v", vector.Post.StateTree.RootCID, actual)
+
+	var (
+		a  = color.New(color.FgMagenta, color.Bold).Sprint("(A) expected final state")
+		b  = color.New(color.FgYellow, color.Bold).Sprint("(B) actual final state")
+		c  = color.New(color.FgCyan, color.Bold).Sprint("(C) initial state")
+		d1 = color.New(color.FgGreen, color.Bold).Sprint("[Δ1]")
+		d2 = color.New(color.FgGreen, color.Bold).Sprint("[Δ2]")
+		d3 = color.New(color.FgGreen, color.Bold).Sprint("[Δ3]")
+	)
+
+	bold := color.New(color.Bold).SprintfFunc()
+
+	// run state diffs.
+	t.Log(bold("=== dumping 3-way diffs between %s, %s, %s ===", a, b, c))
+
+	t.Log(bold("--- %s left: %s; right: %s ---", d1, a, b))
+	t.Log(statediff.Diff(context.Background(), bs, vector.Post.StateTree.RootCID, actual))
+
+	t.Log(bold("--- %s left: %s; right: %s ---", d2, c, b))
+	t.Log(statediff.Diff(context.Background(), bs, vector.Pre.StateTree.RootCID, actual))
+
+	t.Log(bold("--- %s left: %s; right: %s ---", d3, c, a))
+	t.Log(statediff.Diff(context.Background(), bs, vector.Pre.StateTree.RootCID, vector.Post.StateTree.RootCID))
+}
+
+func loadCAR(t *testing.T, vectorCAR schema.Base64EncodedBytes) blockstore.Blockstore {
+	bs := blockstore.NewTemporary()
+
+	// Read the base64-encoded CAR from the vector, and inflate the gzip.
+	buf := bytes.NewReader(vectorCAR)
+	r, err := gzip.NewReader(buf)
+	if err != nil {
+		t.Fatalf("failed to inflate gzipped CAR: %s", err)
+	}
+	defer r.Close() // nolint
+
+	// Load the CAR embedded in the test vector into the Blockstore.
+	_, err = car.LoadCar(bs, r)
+	if err != nil {
+		t.Fatalf("failed to load state tree car from test vector: %s", err)
+	}
+	return bs
+}

documentation/en/api-methods.md

@@ -85,6 +85,9 @@
 * [Net](#Net)
   * [NetAddrsListen](#NetAddrsListen)
   * [NetAutoNatStatus](#NetAutoNatStatus)
+  * [NetBandwidthStats](#NetBandwidthStats)
+  * [NetBandwidthStatsByPeer](#NetBandwidthStatsByPeer)
+  * [NetBandwidthStatsByProtocol](#NetBandwidthStatsByProtocol)
   * [NetConnect](#NetConnect)
   * [NetConnectedness](#NetConnectedness)
   * [NetDisconnect](#NetDisconnect)
@@ -198,7 +201,7 @@ Response:
 ```json
 {
   "Version": "string value",
-  "APIVersion": 3328,
+  "APIVersion": 3584,
   "BlockDelay": 42
 }
 ```
@@ -268,6 +271,9 @@ blockchain, but that do not require any form of state computation.
 
 ### ChainExport
 ChainExport returns a stream of bytes with CAR dump of chain data.
+The exported chain data includes the header chain from the given tipset
+back to genesis, the entire genesis state, and the most recent 'nroots'
+state trees.
 
 
 Perms: read
@@ -275,6 +281,7 @@ Perms: read
 Inputs:
 ```json
 [
+  10101,
   [
     {
       "/": "bafy2bzacea3wsdh6y3a36tb3skempjoxqpuyompjbmfeyf34fi3uy6uue42v4"
@@ -2055,6 +2062,61 @@ Response:
 }
 ```
 
+### NetBandwidthStats
+
+
+Perms: read
+
+Inputs: `null`
+
+Response:
+```json
+{
+  "TotalIn": 9,
+  "TotalOut": 9,
+  "RateIn": 12.3,
+  "RateOut": 12.3
+}
+```
+
+### NetBandwidthStatsByPeer
+
+
+Perms: read
+
+Inputs: `null`
+
+Response:
+```json
+{
+  "12D3KooWSXmXLJmBR1M7i9RW9GQPNUhZSzXKzxDHWtAgNuJAbyEJ": {
+    "TotalIn": 174000,
+    "TotalOut": 12500,
+    "RateIn": 100,
+    "RateOut": 50
+  }
+}
+```
+
+### NetBandwidthStatsByProtocol
+
+
+Perms: read
+
+Inputs: `null`
+
+Response:
+```json
+{
+  "/fil/hello/1.0.0": {
+    "TotalIn": 174000,
+    "TotalOut": 12500,
+    "RateIn": 100,
+    "RateOut": 50
+  }
+}
+```
+
 ### NetConnect
 
 

documentation/en/api.md

@@ -26,7 +26,7 @@ The necessary permissions for each are in [api/struct.go](https://github.com/fil
 
 ## How do I make an API request?
 
-To demonstrate making an API request, we will take the method `ChainHead` from [api/api.go](https://github.com/filecoin-project/lotus/blob/master/api/api_full.go).
+To demonstrate making an API request, we will take the method `ChainHead` from [api/api_full.go](https://github.com/filecoin-project/lotus/blob/master/api/api_full.go).
 
 ```go
 ChainHead(context.Context) (*types.TipSet, error)

documentation/en/mpool.md

@@ -133,7 +133,7 @@ type MpoolConfig struct {
 
 The meaning of these fields is as follows:
 - `PriorityAddrs` -- these are the addresses of actors whose pending messages should always
-  be included in a block during message selection, regardless of profitability.
+  be included in a block during message selection, as long as they are profitable.
   Miners should configure their own worker addresses so that they include their own messages
   when they produce a new block.
   Default is empty.

extern/sector-storage/ffiwrapper/sealer_cgo.go

@@ -371,7 +371,11 @@ func (sb *Sealer) ReadPiece(ctx context.Context, writer io.Writer, sector abi.Se
 	maxPieceSize := abi.PaddedPieceSize(sb.ssize)
 
 	pf, err := openPartialFile(maxPieceSize, path.Unsealed)
-	if xerrors.Is(err, os.ErrNotExist) {
+	if err != nil {
+		if xerrors.Is(err, os.ErrNotExist) {
+			return false, nil
+		}
+
 		return false, xerrors.Errorf("opening partial file: %w", err)
 	}
 

extern/sector-storage/localworker.go

@@ -208,8 +208,8 @@ func (l *LocalWorker) Remove(ctx context.Context, sector abi.SectorID) error {
 	return err
 }
 
-func (l *LocalWorker) MoveStorage(ctx context.Context, sector abi.SectorID) error {
-	if err := l.storage.MoveStorage(ctx, sector, l.scfg.SealProofType, stores.FTSealed|stores.FTCache); err != nil {
+func (l *LocalWorker) MoveStorage(ctx context.Context, sector abi.SectorID, types stores.SectorFileType) error {
+	if err := l.storage.MoveStorage(ctx, sector, l.scfg.SealProofType, types); err != nil {
 		return xerrors.Errorf("moving sealed data to storage: %w", err)
 	}
 

extern/sector-storage/manager.go

@@ -31,7 +31,7 @@ type URLs []string
 type Worker interface {
 	ffiwrapper.StorageSealer
 
-	MoveStorage(ctx context.Context, sector abi.SectorID) error
+	MoveStorage(ctx context.Context, sector abi.SectorID, types stores.SectorFileType) error
 
 	Fetch(ctx context.Context, s abi.SectorID, ft stores.SectorFileType, ptype stores.PathType, am stores.AcquireMode) error
 	UnsealPiece(context.Context, abi.SectorID, storiface.UnpaddedByteIndex, abi.UnpaddedPieceSize, abi.SealRandomness, cid.Cid) error
@@ -441,7 +441,7 @@ func (m *Manager) FinalizeSector(ctx context.Context, sector abi.SectorID, keepU
 	err = m.sched.Schedule(ctx, sector, sealtasks.TTFetch, fetchSel,
 		schedFetch(sector, stores.FTCache|stores.FTSealed|moveUnsealed, stores.PathStorage, stores.AcquireMove),
 		func(ctx context.Context, w Worker) error {
-			return w.MoveStorage(ctx, sector)
+			return w.MoveStorage(ctx, sector, stores.FTCache|stores.FTSealed|moveUnsealed)
 		})
 	if err != nil {
 		return xerrors.Errorf("moving sector to storage: %w", err)

extern/sector-storage/sched.go

@@ -437,6 +437,10 @@ func (sh *scheduler) trySched() {
 			log.Debugf("SCHED ASSIGNED sqi:%d sector %d task %s to window %d", sqi, task.sector.Number, task.taskType, wnd)
 
 			windows[wnd].allocated.add(wr, needRes)
+			// TODO: We probably want to re-sort acceptableWindows here based on new
+			//  workerHandle.utilization + windows[wnd].allocated.utilization (workerHandle.utilization is used in all
+			//  task selectors, but not in the same way, so need to figure out how to do that in a non-O(n^2 way), and
+			//  without additional network roundtrips (O(n^2) could be avoided by turning acceptableWindows.[] into heaps))
 
 			selectedWindow = wnd
 			break
@@ -559,6 +563,7 @@ func (sh *scheduler) runWorker(wid WorkerID) {
 				return
 			}
 
+			sh.workersLk.RLock()
 			worker.wndLk.Lock()
 
 			windowsRequested -= sh.workerCompactWindows(worker, wid)
@@ -570,8 +575,6 @@ func (sh *scheduler) runWorker(wid WorkerID) {
 
 				// process tasks within a window, preferring tasks at lower indexes
 				for len(firstWindow.todo) > 0 {
-					sh.workersLk.RLock()
-
 					tidx := -1
 
 					worker.lk.Lock()
@@ -585,7 +588,6 @@ func (sh *scheduler) runWorker(wid WorkerID) {
 					worker.lk.Unlock()
 
 					if tidx == -1 {
-						sh.workersLk.RUnlock()
 						break assignLoop
 					}
 
@@ -593,7 +595,6 @@ func (sh *scheduler) runWorker(wid WorkerID) {
 
 					log.Debugf("assign worker sector %d", todo.sector.Number)
 					err := sh.assignWorker(taskDone, wid, worker, todo)
-					sh.workersLk.RUnlock()
 
 					if err != nil {
 						log.Error("assignWorker error: %+v", err)
@@ -614,6 +615,7 @@ func (sh *scheduler) runWorker(wid WorkerID) {
 			}
 
 			worker.wndLk.Unlock()
+			sh.workersLk.RUnlock()
 		}
 	}()
 }
@@ -772,14 +774,19 @@ func (sh *scheduler) dropWorker(wid WorkerID) {
 }
 
 func (sh *scheduler) workerCleanup(wid WorkerID, w *workerHandle) {
-	if !w.cleanupStarted {
+	select {
+	case <-w.closingMgr:
+	default:
 		close(w.closingMgr)
 	}
+
+	sh.workersLk.Unlock()
 	select {
 	case <-w.closedMgr:
 	case <-time.After(time.Second):
 		log.Errorf("timeout closing worker manager goroutine %d", wid)
 	}
+	sh.workersLk.Lock()
 
 	if !w.cleanupStarted {
 		w.cleanupStarted = true

extern/sector-storage/sched_resources.go

@@ -108,3 +108,17 @@ func (a *activeResources) utilization(wr storiface.WorkerResources) float64 {
 
 	return max
 }
+
+func (wh *workerHandle) utilization() float64 {
+	wh.lk.Lock()
+	u := wh.active.utilization(wh.info.Resources)
+	u += wh.preparing.utilization(wh.info.Resources)
+	wh.lk.Unlock()
+	wh.wndLk.Lock()
+	for _, window := range wh.activeWindows {
+		u += window.allocated.utilization(wh.info.Resources)
+	}
+	wh.wndLk.Unlock()
+
+	return u
+}

extern/sector-storage/sched_test.go

@@ -82,7 +82,7 @@ func (s *schedTestWorker) AddPiece(ctx context.Context, sector abi.SectorID, pie
 	panic("implement me")
 }
 
-func (s *schedTestWorker) MoveStorage(ctx context.Context, sector abi.SectorID) error {
+func (s *schedTestWorker) MoveStorage(ctx context.Context, sector abi.SectorID, types stores.SectorFileType) error {
 	panic("implement me")
 }
 

extern/sector-storage/selector_alloc.go

@@ -59,7 +59,7 @@ func (s *allocSelector) Ok(ctx context.Context, task sealtasks.TaskType, spt abi
 }
 
 func (s *allocSelector) Cmp(ctx context.Context, task sealtasks.TaskType, a, b *workerHandle) (bool, error) {
-	return a.active.utilization(a.info.Resources) < b.active.utilization(b.info.Resources), nil
+	return a.utilization() < b.utilization(), nil
 }
 
 var _ WorkerSelector = &allocSelector{}

extern/sector-storage/selector_existing.go

@@ -61,7 +61,7 @@ func (s *existingSelector) Ok(ctx context.Context, task sealtasks.TaskType, spt
 }
 
 func (s *existingSelector) Cmp(ctx context.Context, task sealtasks.TaskType, a, b *workerHandle) (bool, error) {
-	return a.active.utilization(a.info.Resources) < b.active.utilization(b.info.Resources), nil
+	return a.utilization() < b.utilization(), nil
 }
 
 var _ WorkerSelector = &existingSelector{}

extern/sector-storage/selector_task.go

@@ -42,7 +42,7 @@ func (s *taskSelector) Cmp(ctx context.Context, _ sealtasks.TaskType, a, b *work
 		return len(atasks) < len(btasks), nil // prefer workers which can do less
 	}
 
-	return a.active.utilization(a.info.Resources) < b.active.utilization(b.info.Resources), nil
+	return a.utilization() < b.utilization(), nil
 }
 
 var _ WorkerSelector = &allocSelector{}

extern/sector-storage/testworker_test.go

@@ -85,7 +85,7 @@ func (t *testWorker) Remove(ctx context.Context, sector abi.SectorID) error {
 	panic("implement me")
 }
 
-func (t *testWorker) MoveStorage(ctx context.Context, sector abi.SectorID) error {
+func (t *testWorker) MoveStorage(ctx context.Context, sector abi.SectorID, types stores.SectorFileType) error {
 	panic("implement me")
 }
 

extern/storage-sealing/fsm.go

@@ -376,12 +376,33 @@ func (m *Sealing) restartSectors(ctx context.Context) error {
 		return xerrors.Errorf("getting the sealing delay: %w", err)
 	}
 
+	m.unsealedInfoMap.lk.Lock()
+	defer m.unsealedInfoMap.lk.Unlock()
 	for _, sector := range trackedSectors {
 		if err := m.sectors.Send(uint64(sector.SectorNumber), SectorRestart{}); err != nil {
 			log.Errorf("restarting sector %d: %+v", sector.SectorNumber, err)
 		}
 
 		if sector.State == WaitDeals {
+
+			// put the sector in the unsealedInfoMap
+			if _, ok := m.unsealedInfoMap.infos[sector.SectorNumber]; ok {
+				// something's funky here, but probably safe to move on
+				log.Warnf("sector %v was already in the unsealedInfoMap when restarting", sector.SectorNumber)
+			} else {
+				ui := UnsealedSectorInfo{}
+				for _, p := range sector.Pieces {
+					if p.DealInfo != nil {
+						ui.numDeals++
+					}
+					ui.stored += p.Piece.Size
+					ui.pieceSizes = append(ui.pieceSizes, p.Piece.Size.Unpadded())
+				}
+
+				m.unsealedInfoMap.infos[sector.SectorNumber] = ui
+			}
+
+			// start a fresh timer for the sector
 			if cfg.WaitDealsDelay > 0 {
 				timer := time.NewTimer(cfg.WaitDealsDelay)
 				go func() {

go.mod

@@ -7,19 +7,18 @@ replace github.com/supranational/blst => github.com/supranational/blst v0.1.2-al
 require (
 	contrib.go.opencensus.io/exporter/jaeger v0.1.0
 	contrib.go.opencensus.io/exporter/prometheus v0.1.0
-	github.com/AndreasBriese/bbloom v0.0.0-20190825152654-46b345b51c96 // indirect
 	github.com/BurntSushi/toml v0.3.1
 	github.com/GeertJohan/go.rice v1.0.0
 	github.com/Gurpartap/async v0.0.0-20180927173644-4f7f499dd9ee
 	github.com/acarl005/stripansi v0.0.0-20180116102854-5a71ef0e047d
 	github.com/buger/goterm v0.0.0-20200322175922-2f3e71b85129
 	github.com/coreos/go-systemd/v22 v22.0.0
-	github.com/davidlazar/go-crypto v0.0.0-20200604182044-b73af7476f6c // indirect
 	github.com/detailyang/go-fallocate v0.0.0-20180908115635-432fa640bd2e
 	github.com/dgraph-io/badger/v2 v2.0.3
 	github.com/docker/go-units v0.4.0
 	github.com/drand/drand v1.0.3-0.20200714175734-29705eaf09d4
 	github.com/drand/kyber v1.1.1
+	github.com/dustin/go-humanize v1.0.0
 	github.com/elastic/go-sysinfo v1.3.0
 	github.com/fatih/color v1.8.0
 	github.com/filecoin-project/chain-validation v0.0.6-0.20200813000554-40c22fe26eef
@@ -30,7 +29,7 @@ require (
 	github.com/filecoin-project/go-crypto v0.0.0-20191218222705-effae4ea9f03
 	github.com/filecoin-project/go-data-transfer v0.6.3
 	github.com/filecoin-project/go-fil-commcid v0.0.0-20200716160307-8f644712406f
-	github.com/filecoin-project/go-fil-markets v0.5.8
+	github.com/filecoin-project/go-fil-markets v0.5.9
 	github.com/filecoin-project/go-jsonrpc v0.1.2-0.20200822201400-474f4fdccc52
 	github.com/filecoin-project/go-multistore v0.0.3
 	github.com/filecoin-project/go-padreader v0.0.0-20200210211231-548257017ca6
@@ -41,7 +40,7 @@ require (
 	github.com/filecoin-project/specs-actors v0.9.3
 	github.com/filecoin-project/specs-storage v0.1.1-0.20200730063404-f7db367e9401
 	github.com/filecoin-project/statediff v0.0.1
-	github.com/filecoin-project/test-vectors v0.0.0-20200826113833-9ffe6524729d
+	github.com/filecoin-project/test-vectors v0.0.0-20200902131127-9806d09b005d
 	github.com/gbrlsnchs/jwt/v3 v3.0.0-beta.1
 	github.com/go-kit/kit v0.10.0
 	github.com/google/uuid v1.1.1
@@ -92,7 +91,7 @@ require (
 	github.com/libp2p/go-libp2p-mplex v0.2.4
 	github.com/libp2p/go-libp2p-noise v0.1.1
 	github.com/libp2p/go-libp2p-peerstore v0.2.6
-	github.com/libp2p/go-libp2p-pubsub v0.3.5
+	github.com/libp2p/go-libp2p-pubsub v0.3.6-0.20200901174250-06a12f17b7de
 	github.com/libp2p/go-libp2p-quic-transport v0.8.0
 	github.com/libp2p/go-libp2p-record v0.1.3
 	github.com/libp2p/go-libp2p-routing-helpers v0.2.3
@@ -119,7 +118,6 @@ require (
 	github.com/whyrusleeping/pubsub v0.0.0-20131020042734-02de8aa2db3d
 	github.com/xorcare/golden v0.6.1-0.20191112154924-b87f686d7542
 	go.opencensus.io v0.22.4
-	go.uber.org/dig v1.10.0 // indirect
 	go.uber.org/fx v1.9.0
 	go.uber.org/multierr v1.5.0
 	go.uber.org/zap v1.15.0

go.sum

@@ -33,9 +33,8 @@ dmitri.shuralyov.com/service/change v0.0.0-20181023043359-a85b471d5412/go.mod h1
 dmitri.shuralyov.com/state v0.0.0-20180228185332-28bcc343414c/go.mod h1:0PRwlb0D6DFvNNtx+9ybjezNCa8XF0xaYcETyp6rHWU=
 git.apache.org/thrift.git v0.0.0-20180902110319-2566ecd5d999/go.mod h1:fPE2ZNJGynbRyZ4dJvy6G277gSllfV2HJqblrnkyeyg=
 github.com/AndreasBriese/bbloom v0.0.0-20180913140656-343706a395b7/go.mod h1:bOvUY6CB00SOBii9/FifXqc0awNKxLFCL/+pkDPuyl8=
+github.com/AndreasBriese/bbloom v0.0.0-20190306092124-e2d15f34fcf9 h1:HD8gA2tkByhMAwYaFAX9w2l7vxvBQ5NMoxDrkhqhtn4=
 github.com/AndreasBriese/bbloom v0.0.0-20190306092124-e2d15f34fcf9/go.mod h1:bOvUY6CB00SOBii9/FifXqc0awNKxLFCL/+pkDPuyl8=
-github.com/AndreasBriese/bbloom v0.0.0-20190825152654-46b345b51c96 h1:cTp8I5+VIoKjsnZuH8vjyaysT/ses3EvZeaV/1UkF2M=
-github.com/AndreasBriese/bbloom v0.0.0-20190825152654-46b345b51c96/go.mod h1:bOvUY6CB00SOBii9/FifXqc0awNKxLFCL/+pkDPuyl8=
 github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo=
@@ -163,9 +162,8 @@ github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davidlazar/go-crypto v0.0.0-20170701192655-dcfb0a7ac018/go.mod h1:rQYf4tfk5sSwFsnDg3qYaBxSjsD9S8+59vW0dKUgme4=
+github.com/davidlazar/go-crypto v0.0.0-20190912175916-7055855a373f h1:BOaYiTvg8p9vBUXpklC22XSK/mifLF7lG9jtmYYi3Tc=
 github.com/davidlazar/go-crypto v0.0.0-20190912175916-7055855a373f/go.mod h1:rQYf4tfk5sSwFsnDg3qYaBxSjsD9S8+59vW0dKUgme4=
-github.com/davidlazar/go-crypto v0.0.0-20200604182044-b73af7476f6c h1:pFUpOrbxDR6AkioZ1ySsx5yxlDQZ8stG2b88gTPxgJU=
-github.com/davidlazar/go-crypto v0.0.0-20200604182044-b73af7476f6c/go.mod h1:6UhI8N9EjYm1c2odKpFpAYeR8dsBeM7PtzQhRgxRr9U=
 github.com/detailyang/go-fallocate v0.0.0-20180908115635-432fa640bd2e h1:lj77EKYUpYXTd8CD/+QMIf8b6OIOTsfEBSXiAzuEHTU=
 github.com/detailyang/go-fallocate v0.0.0-20180908115635-432fa640bd2e/go.mod h1:3ZQK6DMPSz/QZ73jlWxBtUhNA8xZx7LzUFSq/OfP8vk=
 github.com/dgraph-io/badger v1.5.5-0.20190226225317-8115aed38f8f/go.mod h1:VZxzAIRPHRVNRKRo6AXrX9BJegn6il06VMTZVJYCIjQ=
@@ -246,9 +244,10 @@ github.com/filecoin-project/go-fil-commcid v0.0.0-20200208005934-2b8bd03caca5/go
 github.com/filecoin-project/go-fil-commcid v0.0.0-20200716160307-8f644712406f h1:GxJzR3oRIMTPtpZ0b7QF8FKPK6/iPAc7trhlL5k/g+s=
 github.com/filecoin-project/go-fil-commcid v0.0.0-20200716160307-8f644712406f/go.mod h1:Eaox7Hvus1JgPrL5+M3+h7aSPHc0cVqpSxA+TxIEpZQ=
 github.com/filecoin-project/go-fil-markets v0.5.6-0.20200814234959-80b1788108ac/go.mod h1:umicPCaN99ysHTiYOmwhuLxTFbOwcsI+mdw/t96vvM4=
-github.com/filecoin-project/go-fil-markets v0.5.6/go.mod h1:SJApXAKr5jyGpbzDEOhvemui0pih7hhT8r2MXJxCP1E=
 github.com/filecoin-project/go-fil-markets v0.5.8 h1:uwl0QNUVmmSlUQfxshpj21Dmhh6WKTQNhnb1GMfdp18=
 github.com/filecoin-project/go-fil-markets v0.5.8/go.mod h1:6ZX1vbZbnukbVQ8tCB/MmEizuW/bmRX7SpGAltU3KVg=
+github.com/filecoin-project/go-fil-markets v0.5.9 h1:iIO17UfIjUCiB37TRwgiBwAyfJJwHb8e8uAfu7F37gc=
+github.com/filecoin-project/go-fil-markets v0.5.9/go.mod h1:/cb1IoaiHhwFEWyIAPm9yN6Z+MiPujFZBT8BGH7LwB8=
 github.com/filecoin-project/go-jsonrpc v0.1.2-0.20200817153016-2ea5cbaf5ec0/go.mod h1:XBBpuKIMaXIIzeqzO1iucq4GvbF8CxmXRFoezRh+Cx4=
 github.com/filecoin-project/go-jsonrpc v0.1.2-0.20200822201400-474f4fdccc52 h1:FXtCp0ybqdQL9knb3OGDpkNTaBbPxgkqPeWKotUwkH0=
 github.com/filecoin-project/go-jsonrpc v0.1.2-0.20200822201400-474f4fdccc52/go.mod h1:XBBpuKIMaXIIzeqzO1iucq4GvbF8CxmXRFoezRh+Cx4=
@@ -268,8 +267,8 @@ github.com/filecoin-project/go-statestore v0.1.0 h1:t56reH59843TwXHkMcwyuayStBIi
 github.com/filecoin-project/go-statestore v0.1.0/go.mod h1:LFc9hD+fRxPqiHiaqUEZOinUJB4WARkRfNl10O7kTnI=
 github.com/filecoin-project/go-storedcounter v0.0.0-20200421200003-1c99c62e8a5b h1:fkRZSPrYpk42PV3/lIXiL0LHetxde7vyYYvSsttQtfg=
 github.com/filecoin-project/go-storedcounter v0.0.0-20200421200003-1c99c62e8a5b/go.mod h1:Q0GQOBtKf1oE10eSXSlhN45kDBdGvEcVOqMiffqX+N8=
-github.com/filecoin-project/lotus v0.4.3-0.20200819134055-b13681df3205/go.mod h1:rooripL/X8ixwUngDPzphAv/RKZXWBprbyxxDW0EJi0=
 github.com/filecoin-project/lotus v0.4.3-0.20200820203717-d1718369a182/go.mod h1:biFZPQ/YyQGfkHUmHMiaNf2hnD6zm1+OAXPQYQ61Zkg=
+github.com/filecoin-project/lotus v0.5.8-0.20200902130912-0962292f920e/go.mod h1:OkZ5aUqs+fFnJOq9243WJDsTa9c3/Ae67NIAwVhAB+0=
 github.com/filecoin-project/sector-storage v0.0.0-20200712023225-1d67dcfa3c15/go.mod h1:salgVdX7qeXFo/xaiEQE29J4pPkjn71T0kt0n+VDBzo=
 github.com/filecoin-project/sector-storage v0.0.0-20200730050024-3ee28c3b6d9a/go.mod h1:oOawOl9Yk+qeytLzzIryjI8iRbqo+qzS6EEeElP4PWA=
 github.com/filecoin-project/sector-storage v0.0.0-20200810171746-eac70842d8e0 h1:E1fZ27fhKK05bhZItfTwqr1i05vXnEZJznQFEYwEEUU=
@@ -691,6 +690,8 @@ github.com/joeshaw/multierror v0.0.0-20140124173710-69b34d4ec901/go.mod h1:Z86h9
 github.com/jonboulle/clockwork v0.1.0/go.mod h1:Ii8DK3G1RaLaWxj9trq07+26W01tbo22gdxWY5EU2bo=
 github.com/jonboulle/clockwork v0.1.1-0.20190114141812-62fb9bc030d1 h1:qBCV/RLV02TSfQa7tFmxTihnG+u+7JXByOkhlkR5rmQ=
 github.com/jonboulle/clockwork v0.1.1-0.20190114141812-62fb9bc030d1/go.mod h1:Ii8DK3G1RaLaWxj9trq07+26W01tbo22gdxWY5EU2bo=
+github.com/jpillora/backoff v1.0.0 h1:uvFg412JmmHBHw7iwprIxkPMI+sGQ4kzOWsMeHnm2EA=
+github.com/jpillora/backoff v1.0.0/go.mod h1:J/6gKK9jxlEcS3zixgDgUAsiuZ7yrSoa/FX5e0EB2j4=
 github.com/jrick/logrotate v1.0.0/go.mod h1:LNinyqDIJnpAur+b8yyulnQw/wDuN1+BYKlTRt3OuAQ=
 github.com/jsimonetti/rtnetlink v0.0.0-20190606172950-9527aa82566a/go.mod h1:Oz+70psSo5OFh8DBl0Zv2ACw7Esh6pPUphlvZG9x7uw=
 github.com/jsimonetti/rtnetlink v0.0.0-20190830100107-3784a6c7c552/go.mod h1:Oz+70psSo5OFh8DBl0Zv2ACw7Esh6pPUphlvZG9x7uw=
@@ -828,6 +829,7 @@ github.com/libp2p/go-libp2p-core v0.6.1 h1:XS+Goh+QegCDojUZp00CaPMfiEADCrLjNZskW
 github.com/libp2p/go-libp2p-core v0.6.1/go.mod h1:FfewUH/YpvWbEB+ZY9AQRQ4TAD8sJBt/G1rVvhz5XT8=
 github.com/libp2p/go-libp2p-crypto v0.0.1/go.mod h1:yJkNyDmO341d5wwXxDUGO0LykUVT72ImHNUqh5D/dBE=
 github.com/libp2p/go-libp2p-crypto v0.0.2/go.mod h1:eETI5OUfBnvARGOHrJz2eWNyTUxEGZnBxMcbUjfIj4I=
+github.com/libp2p/go-libp2p-crypto v0.1.0 h1:k9MFy+o2zGDNGsaoZl0MA3iZ75qXxr9OOoAZF+sD5OQ=
 github.com/libp2p/go-libp2p-crypto v0.1.0/go.mod h1:sPUokVISZiy+nNuTTH/TY+leRSxnFj/2GLjtOTW90hI=
 github.com/libp2p/go-libp2p-daemon v0.2.2/go.mod h1:kyrpsLB2JeNYR2rvXSVWyY0iZuRIMhqzWR3im9BV6NQ=
 github.com/libp2p/go-libp2p-discovery v0.0.1/go.mod h1:ZkkF9xIFRLA1xCc7bstYFkd80gBGK8Fc1JqGoU2i+zI=
@@ -853,6 +855,7 @@ github.com/libp2p/go-libp2p-kbucket v0.4.2/go.mod h1:7sCeZx2GkNK1S6lQnGUW5JYZCFP
 github.com/libp2p/go-libp2p-loggables v0.0.1/go.mod h1:lDipDlBNYbpyqyPX/KcoO+eq0sJYEVR2JgOexcivchg=
 github.com/libp2p/go-libp2p-loggables v0.1.0 h1:h3w8QFfCt2UJl/0/NW4K829HX/0S4KD31PQ7m8UXXO8=
 github.com/libp2p/go-libp2p-loggables v0.1.0/go.mod h1:EyumB2Y6PrYjr55Q3/tiJ/o3xoDasoRYM7nOzEpoa90=
+github.com/libp2p/go-libp2p-metrics v0.0.1 h1:yumdPC/P2VzINdmcKZd0pciSUCpou+s0lwYCjBbzQZU=
 github.com/libp2p/go-libp2p-metrics v0.0.1/go.mod h1:jQJ95SXXA/K1VZi13h52WZMa9ja78zjyy5rspMsC/08=
 github.com/libp2p/go-libp2p-mplex v0.1.1/go.mod h1:KUQWpGkCzfV7UIpi8SKsAVxyBgz1c9R5EvxgnwLsb/I=
 github.com/libp2p/go-libp2p-mplex v0.2.0/go.mod h1:Ejl9IyjvXJ0T9iqUTE1jpYATQ9NM3g+OtR+EMMODbKo=
@@ -875,6 +878,7 @@ github.com/libp2p/go-libp2p-noise v0.1.1 h1:vqYQWvnIcHpIoWJKC7Al4D6Hgj0H012TuXRh
 github.com/libp2p/go-libp2p-noise v0.1.1/go.mod h1:QDFLdKX7nluB7DEnlVPbz7xlLHdwHFA9HiohJRr3vwM=
 github.com/libp2p/go-libp2p-peer v0.0.1/go.mod h1:nXQvOBbwVqoP+T5Y5nCjeH4sP9IX/J0AMzcDUVruVoo=
 github.com/libp2p/go-libp2p-peer v0.1.1/go.mod h1:jkF12jGB4Gk/IOo+yomm+7oLWxF278F7UnrYUQ1Q8es=
+github.com/libp2p/go-libp2p-peer v0.2.0 h1:EQ8kMjaCUwt/Y5uLgjT8iY2qg0mGUT0N1zUjer50DsY=
 github.com/libp2p/go-libp2p-peer v0.2.0/go.mod h1:RCffaCvUyW2CJmG2gAWVqwePwW7JMgxjsHm7+J5kjWY=
 github.com/libp2p/go-libp2p-peerstore v0.0.1/go.mod h1:RabLyPVJLuNQ+GFyoEkfi8H4Ti6k/HtZJ7YKgtSq+20=
 github.com/libp2p/go-libp2p-peerstore v0.0.6/go.mod h1:RabLyPVJLuNQ+GFyoEkfi8H4Ti6k/HtZJ7YKgtSq+20=
@@ -891,13 +895,13 @@ github.com/libp2p/go-libp2p-peerstore v0.2.6/go.mod h1:ss/TWTgHZTMpsU/oKVVPQCGuD
 github.com/libp2p/go-libp2p-pnet v0.2.0 h1:J6htxttBipJujEjz1y0a5+eYoiPcFHhSYHH6na5f0/k=
 github.com/libp2p/go-libp2p-pnet v0.2.0/go.mod h1:Qqvq6JH/oMZGwqs3N1Fqhv8NVhrdYcO0BW4wssv21LA=
 github.com/libp2p/go-libp2p-protocol v0.0.1/go.mod h1:Af9n4PiruirSDjHycM1QuiMi/1VZNHYcK8cLgFJLZ4s=
+github.com/libp2p/go-libp2p-protocol v0.1.0 h1:HdqhEyhg0ToCaxgMhnOmUO8snQtt/kQlcjVk3UoJU3c=
 github.com/libp2p/go-libp2p-protocol v0.1.0/go.mod h1:KQPHpAabB57XQxGrXCNvbL6UEXfQqUgC/1adR2Xtflk=
 github.com/libp2p/go-libp2p-pubsub v0.1.1/go.mod h1:ZwlKzRSe1eGvSIdU5bD7+8RZN/Uzw0t1Bp9R1znpR/Q=
 github.com/libp2p/go-libp2p-pubsub v0.3.2-0.20200527132641-c0712c6e92cf/go.mod h1:TxPOBuo1FPdsTjFnv+FGZbNbWYsp74Culx+4ViQpato=
-github.com/libp2p/go-libp2p-pubsub v0.3.4/go.mod h1:DTMSVmZZfXodB/pvdTGrY2eHPZ9W2ev7hzTH83OKHrI=
 github.com/libp2p/go-libp2p-pubsub v0.3.5-0.20200820194335-bfc96c2cd081/go.mod h1:DTMSVmZZfXodB/pvdTGrY2eHPZ9W2ev7hzTH83OKHrI=
-github.com/libp2p/go-libp2p-pubsub v0.3.5 h1:iF75GWpcxKEUQU8tTkgLy69qIQvfhL+t6U6ndQrB6ho=
-github.com/libp2p/go-libp2p-pubsub v0.3.5/go.mod h1:DTMSVmZZfXodB/pvdTGrY2eHPZ9W2ev7hzTH83OKHrI=
+github.com/libp2p/go-libp2p-pubsub v0.3.6-0.20200901174250-06a12f17b7de h1:Dl0B0x6u+OSKXAa1DeB6xHFsUOBAhjrXJ10zykVSN6Q=
+github.com/libp2p/go-libp2p-pubsub v0.3.6-0.20200901174250-06a12f17b7de/go.mod h1:DTMSVmZZfXodB/pvdTGrY2eHPZ9W2ev7hzTH83OKHrI=
 github.com/libp2p/go-libp2p-quic-transport v0.1.1/go.mod h1:wqG/jzhF3Pu2NrhJEvE+IE0NTHNXslOPn9JQzyCAxzU=
 github.com/libp2p/go-libp2p-quic-transport v0.5.0/go.mod h1:IEcuC5MLxvZ5KuHKjRu+dr3LjCT1Be3rcD/4d8JrX8M=
 github.com/libp2p/go-libp2p-quic-transport v0.7.1/go.mod h1:TD31to4E5exogR/GWHClXCfkktigjAl5rXSt7HoxNvY=
@@ -1497,9 +1501,8 @@ go.uber.org/atomic v1.5.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ=
 go.uber.org/atomic v1.5.1/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ=
 go.uber.org/atomic v1.6.0 h1:Ezj3JGmsOnG1MoRWQkPBsKLe9DwWD9QeXzTRzzldNVk=
 go.uber.org/atomic v1.6.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ=
+go.uber.org/dig v1.8.0 h1:1rR6hnL/bu1EVcjnRDN5kx1vbIjEJDTGhSQ2B3ddpcI=
 go.uber.org/dig v1.8.0/go.mod h1:X34SnWGr8Fyla9zQNO2GSO2D+TIuqB14OS8JhYocIyw=
-go.uber.org/dig v1.10.0 h1:yLmDDj9/zuDjv3gz8GQGviXMs9TfysIUMUilCpgzUJY=
-go.uber.org/dig v1.10.0/go.mod h1:X34SnWGr8Fyla9zQNO2GSO2D+TIuqB14OS8JhYocIyw=
 go.uber.org/fx v1.9.0 h1:7OAz8ucp35AU8eydejpYG7QrbE8rLKzGhHbZlJi5LYY=
 go.uber.org/fx v1.9.0/go.mod h1:mFdUyAUuJ3w4jAckiKSKbldsxy1ojpAMJ+dVZg5Y0Aw=
 go.uber.org/goleak v1.0.0 h1:qsup4IcBdlmsnGfqyLl4Ntn3C2XCCuKAE7DwHpScyUo=
@@ -1551,7 +1554,6 @@ golang.org/x/crypto v0.0.0-20200317142112-1b76d66859c6/go.mod h1:LzIPMQfyMNhhGPh
 golang.org/x/crypto v0.0.0-20200423211502-4bdfaf469ed5/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20200427165652-729f1e841bcc/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20200510223506-06a226fb4e37/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
-golang.org/x/crypto v0.0.0-20200602180216-279210d13fed/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20200604202706-70a84ac30bf9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9 h1:psW17arqaxU48Z5kZ0CQnkZWQJsqcURM6tKiBApRjXI=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=

node/builder.go

@@ -86,6 +86,7 @@ var (
 	NatPortMapKey        = special{8}  // Libp2p option
 	ConnectionManagerKey = special{9}  // Libp2p option
 	AutoNATSvcKey        = special{10} // Libp2p option
+	BandwidthReporterKey = special{11} // Libp2p option
 )
 
 type invoke int
@@ -181,6 +182,7 @@ func libp2p() Option {
 		Override(new(routing.Routing), lp2p.Routing),
 
 		Override(NatPortMapKey, lp2p.NatPortMap),
+		Override(BandwidthReporterKey, lp2p.BandwidthCounter),
 
 		Override(ConnectionManagerKey, lp2p.ConnectionManager(50, 200, 20*time.Second, nil)),
 		Override(AutoNATSvcKey, lp2p.AutoNATService),

node/impl/common/common.go

@@ -9,8 +9,10 @@ import (
 
 	"github.com/gbrlsnchs/jwt/v3"
 	"github.com/libp2p/go-libp2p-core/host"
+	metrics "github.com/libp2p/go-libp2p-core/metrics"
 	"github.com/libp2p/go-libp2p-core/network"
 	"github.com/libp2p/go-libp2p-core/peer"
+	protocol "github.com/libp2p/go-libp2p-core/protocol"
 	swarm "github.com/libp2p/go-libp2p-swarm"
 	basichost "github.com/libp2p/go-libp2p/p2p/host/basic"
 	ma "github.com/multiformats/go-multiaddr"
@@ -32,6 +34,7 @@ type CommonAPI struct {
 	RawHost      lp2p.RawHost
 	Host         host.Host
 	Router       lp2p.BaseIpfsRouting
+	Reporter     metrics.Reporter
 	Sk           *dtypes.ScoreKeeper
 	ShutdownChan dtypes.ShutdownChan
 }
@@ -133,6 +136,22 @@ func (a *CommonAPI) NetAutoNatStatus(ctx context.Context) (i api.NatInfo, err er
 	}, nil
 }
 
+func (a *CommonAPI) NetBandwidthStats(ctx context.Context) (metrics.Stats, error) {
+	return a.Reporter.GetBandwidthTotals(), nil
+}
+
+func (a *CommonAPI) NetBandwidthStatsByPeer(ctx context.Context) (map[string]metrics.Stats, error) {
+	out := make(map[string]metrics.Stats)
+	for p, s := range a.Reporter.GetBandwidthByPeer() {
+		out[p.String()] = s
+	}
+	return out, nil
+}
+
+func (a *CommonAPI) NetBandwidthStatsByProtocol(ctx context.Context) (map[protocol.ID]metrics.Stats, error) {
+	return a.Reporter.GetBandwidthByProtocol(), nil
+}
+
 func (a *CommonAPI) ID(context.Context) (peer.ID, error) {
 	return a.Host.ID(), nil
 }

node/impl/full/chain.go

@@ -494,7 +494,7 @@ func (a *ChainAPI) ChainGetMessage(ctx context.Context, mc cid.Cid) (*types.Mess
 	return cm.VMMessage(), nil
 }
 
-func (a *ChainAPI) ChainExport(ctx context.Context, tsk types.TipSetKey) (<-chan []byte, error) {
+func (a *ChainAPI) ChainExport(ctx context.Context, nroots abi.ChainEpoch, tsk types.TipSetKey) (<-chan []byte, error) {
 	ts, err := a.Chain.GetTipSetFromKey(tsk)
 	if err != nil {
 		return nil, xerrors.Errorf("loading tipset %s: %w", tsk, err)
@@ -503,7 +503,7 @@ func (a *ChainAPI) ChainExport(ctx context.Context, tsk types.TipSetKey) (<-chan
 	out := make(chan []byte)
 	go func() {
 		defer w.Close() //nolint:errcheck // it is a pipe
-		if err := a.Chain.Export(ctx, ts, w); err != nil {
+		if err := a.Chain.Export(ctx, ts, nroots, w); err != nil {
 			log.Errorf("chain export call failed: %s", err)
 			return
 		}

node/impl/full/gas.go

@@ -14,6 +14,7 @@ import (
 	"github.com/filecoin-project/lotus/chain/types"
 
 	"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/specs-actors/actors/builtin"
 	"github.com/filecoin-project/specs-actors/actors/runtime/exitcode"
@@ -36,27 +37,14 @@ func (a *GasAPI) GasEstimateFeeCap(ctx context.Context, msg *types.Message, maxq
 	tsk types.TipSetKey) (types.BigInt, error) {
 	ts := a.Chain.GetHeaviestTipSet()
 
-	var act types.Actor
-	err := a.Stmgr.WithParentState(ts, a.Stmgr.WithActor(msg.From, stmgr.GetActor(&act)))
-	if err != nil {
-		return types.NewInt(0), xerrors.Errorf("getting actor: %w", err)
-	}
-
 	parentBaseFee := ts.Blocks()[0].ParentBaseFee
 	increaseFactor := math.Pow(1.+1./float64(build.BaseFeeMaxChangeDenom), float64(maxqueueblks))
 
 	feeInFuture := types.BigMul(parentBaseFee, types.NewInt(uint64(increaseFactor*(1<<8))))
-	feeInFuture = types.BigDiv(feeInFuture, types.NewInt(1<<8))
+	out := types.BigDiv(feeInFuture, types.NewInt(1<<8))
 
-	gasLimitBig := types.NewInt(uint64(msg.GasLimit))
-	maxAccepted := types.BigDiv(act.Balance, types.NewInt(MaxSpendOnFeeDenom))
-	expectedFee := types.BigMul(feeInFuture, gasLimitBig)
-
-	out := feeInFuture
-	if types.BigCmp(expectedFee, maxAccepted) > 0 {
-		log.Warnf("Expected fee for message higher than tolerance: %s > %s, setting to tolerance",
-			types.FIL(expectedFee), types.FIL(maxAccepted))
-		out = types.BigDiv(maxAccepted, gasLimitBig)
+	if msg.GasPremium != types.EmptyInt {
+		out = types.BigAdd(out, msg.GasPremium)
 	}
 
 	return out, nil
@@ -109,27 +97,22 @@ func (a *GasAPI) GasEstimateGasPremium(ctx context.Context, nblocksincl uint64,
 		return prices[i].price.GreaterThan(prices[j].price)
 	})
 
-	// todo: account for how full blocks are
-
 	at := build.BlockGasTarget * int64(blocks) / 2
-	prev := big.Zero()
-
-	premium := big.Zero()
+	prev1, prev2 := big.Zero(), big.Zero()
 	for _, price := range prices {
+		prev1, prev2 = price.price, prev1
 		at -= price.limit
 		if at > 0 {
-			prev = price.price
 			continue
 		}
-
-		if prev.Equals(big.Zero()) {
-			return types.BigAdd(price.price, big.NewInt(1)), nil
-		}
-
-		premium = types.BigAdd(big.Div(types.BigAdd(price.price, prev), types.NewInt(2)), big.NewInt(1))
 	}
 
-	if types.BigCmp(premium, big.Zero()) == 0 {
+	premium := prev1
+	if prev2.Sign() != 0 {
+		premium = big.Div(types.BigAdd(prev1, prev2), types.NewInt(2))
+	}
+
+	if types.BigCmp(premium, types.NewInt(MinGasPremium)) < 0 {
 		switch nblocksincl {
 		case 1:
 			premium = types.NewInt(2 * MinGasPremium)
@@ -144,7 +127,7 @@ func (a *GasAPI) GasEstimateGasPremium(ctx context.Context, nblocksincl uint64,
 	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.BigMul(premium, types.NewInt(uint64(noise*(1<<precision))+1))
 	premium = types.BigDiv(premium, types.NewInt(1<<precision))
 	return premium, nil
 }
@@ -219,10 +202,31 @@ func (a *GasAPI) GasEstimateMessageGas(ctx context.Context, msg *types.Message,
 		if err != nil {
 			return nil, xerrors.Errorf("estimating fee cap: %w", err)
 		}
-		msg.GasFeeCap = big.Add(feeCap, msg.GasPremium)
+		msg.GasFeeCap = feeCap
 	}
 
 	capGasFee(msg, spec.Get().MaxFee)
 
 	return msg, nil
 }
+
+func capGasFee(msg *types.Message, maxFee abi.TokenAmount) {
+	if maxFee.Equals(big.Zero()) {
+		maxFee = types.NewInt(build.FilecoinPrecision / 10)
+	}
+
+	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)
+}

node/impl/full/mpool.go

@@ -8,9 +8,6 @@ import (
 	"go.uber.org/fx"
 	"golang.org/x/xerrors"
 
-	"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"
@@ -115,27 +112,6 @@ func (a *MpoolAPI) MpoolPush(ctx context.Context, smsg *types.SignedMessage) (ci
 	return a.Mpool.Push(smsg)
 }
 
-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)

node/modules/chain.go

@@ -38,7 +38,9 @@ func ChainExchange(mctx helpers.MetricsCtx, lc fx.Lifecycle, host host.Host, rt
 	// (so bitswap uses /chain/ipfs/bitswap/1.0.0 internally for chain sync stuff)
 	bitswapNetwork := network.NewFromIpfsHost(host, rt, network.Prefix("/chain"))
 	bitswapOptions := []bitswap.Option{bitswap.ProvideEnabled(false)}
-	exch := bitswap.New(helpers.LifecycleCtx(mctx, lc), bitswapNetwork, bs, bitswapOptions...)
+
+	// Use just exch.Close(), closing the context is not needed
+	exch := bitswap.New(mctx, bitswapNetwork, bs, bitswapOptions...)
 	lc.Append(fx.Hook{
 		OnStop: func(ctx context.Context) error {
 			return exch.Close()

node/modules/storageminer.go

@@ -275,13 +275,14 @@ func StagingDAG(mctx helpers.MetricsCtx, lc fx.Lifecycle, ibs dtypes.StagingBloc
 
 	bitswapNetwork := network.NewFromIpfsHost(h, rt)
 	bitswapOptions := []bitswap.Option{bitswap.ProvideEnabled(false)}
-	exch := bitswap.New(helpers.LifecycleCtx(mctx, lc), bitswapNetwork, ibs, bitswapOptions...)
+	exch := bitswap.New(mctx, bitswapNetwork, ibs, bitswapOptions...)
 
 	bsvc := blockservice.New(ibs, exch)
 	dag := merkledag.NewDAGService(bsvc)
 
 	lc.Append(fx.Hook{
 		OnStop: func(_ context.Context) error {
+			// blockservice closes the exchange
 			return bsvc.Close()
 		},
 	})

node/repo/fsrepo_ds.go

@@ -17,7 +17,7 @@ import (
 type dsCtor func(path string) (datastore.Batching, error)
 
 var fsDatastores = map[string]dsCtor{
-	"chain":    badgerDs,
+	"chain":    chainBadgerDs,
 	"metadata": levelDs,
 
 	// Those need to be fast for large writes... but also need a really good GC :c
@@ -26,6 +26,16 @@ var fsDatastores = map[string]dsCtor{
 	"client": badgerDs, // client specific
 }
 
+func chainBadgerDs(path string) (datastore.Batching, error) {
+	opts := badger.DefaultOptions
+	opts.GcInterval = 0 // disable GC for chain datastore
+
+	opts.Options = dgbadger.DefaultOptions("").WithTruncate(true).
+		WithValueThreshold(1 << 10)
+
+	return badger.NewDatastore(path, &opts)
+}
+
 func badgerDs(path string) (datastore.Batching, error) {
 	opts := badger.DefaultOptions
 	opts.Options = dgbadger.DefaultOptions("").WithTruncate(true).