track expected nonce in mpool, ignore messages with large nonce gaps

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,30 +182,65 @@ 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)
 
 	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,14 +534,48 @@ 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()
 
 	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_test.go

@@ -369,6 +369,12 @@ func TestMessageChainSkipping(t *testing.T) {
 }
 
 func TestBasicMessageSelection(t *testing.T) {
+	oldMaxNonceGap := MaxNonceGap
+	MaxNonceGap = 1000
+	defer func() {
+		MaxNonceGap = oldMaxNonceGap
+	}()
+
 	mp, tma := makeTestMpool()
 
 	// the actors

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: