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

CHANGELOG.md

@@ -1,5 +1,25 @@
 # 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.

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/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"`
@@ -371,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)

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.8"
+const BuildVersion = "0.5.10"
 
 func UserVersion() string {
 	return BuildVersion + buildType() + CurrentCommit

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
 	}
 }
 
@@ -476,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()
@@ -483,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)
@@ -501,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
@@ -664,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
@@ -688,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
 	}
 }
 
@@ -771,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
 		}
 
@@ -780,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) {
@@ -1082,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
@@ -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

@@ -585,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)
@@ -672,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)
 	}
 
@@ -776,7 +778,7 @@ func (mc *msgChain) Before(other *msgChain) 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 || mc.gasPerf < 0) {
-		gasReward := mp.getGasReward(mc.msgs[i], baseFee, ts)
+		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/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/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!")

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)
@@ -2059,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
 
 

go.mod

@@ -18,6 +18,7 @@ require (
 	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

go.sum

@@ -829,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=
@@ -854,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=
@@ -876,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=
@@ -892,6 +895,7 @@ 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=

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/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,28 +37,11 @@ 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))
-
-	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)
-	}
+	out := types.BigDiv(feeInFuture, types.NewInt(1<<8))
 
 	if msg.GasPremium != types.EmptyInt {
 		out = types.BigAdd(out, msg.GasPremium)
@@ -225,3 +209,24 @@ func (a *GasAPI) GasEstimateMessageGas(ctx context.Context, msg *types.Message,
 
 	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)