feat(mempool): priority nonce mempool option with tx replacement

types/mempool/priority_nonce.go

@@ -29,6 +29,7 @@ type priorityNonceMempool struct {
 	senderIndices  map[string]*skiplist.SkipList
 	scores         map[txMeta]txMeta
 	onRead         func(tx sdk.Tx)
+	txReplacement  func(oldTxPriority, newTxPriority int64) bool
 	maxTx          int
 }
 
@@ -92,6 +93,14 @@ func PriorityNonceWithOnRead(onRead func(tx sdk.Tx)) PriorityNonceMempoolOption
 	}
 }
 
+// PriorityNonceWithTxReplacement sets a callback to be called when deplicated tx nonce detected
+// during the mempool insert, the dev can define the tx replacement rule based on the tx priority
+func PriorityNonceWithTxReplacement(txReplacementRule func(oldTxPriority, newTxPriority int64) bool) PriorityNonceMempoolOption {
+	return func(mp *priorityNonceMempool) {
+		mp.txReplacement = txReplacementRule
+	}
+}
+
 // PriorityNonceWithMaxTx sets the maximum number of transactions allowed in the mempool with the semantics:
 //
 // <0: disabled, `Insert` is a no-op
@@ -166,12 +175,6 @@ func (mp *priorityNonceMempool) Insert(ctx context.Context, tx sdk.Tx) error {
 		mp.senderIndices[sender] = senderIndex
 	}
 
-	mp.priorityCounts[priority]++
-
-	// Since senderIndex is scored by nonce, a changed priority will overwrite the
-	// existing key.
-	key.senderElement = senderIndex.Set(key, tx)
-
 	// Since mp.priorityIndex is scored by priority, then sender, then nonce, a
 	// changed priority will create a new key, so we must remove the old key and
 	// re-insert it to avoid having the same tx with different priorityIndex indexed
@@ -181,6 +184,10 @@ func (mp *priorityNonceMempool) Insert(ctx context.Context, tx sdk.Tx) error {
 	// changes.
 	sk := txMeta{nonce: nonce, sender: sender}
 	if oldScore, txExists := mp.scores[sk]; txExists {
+		if mp.txReplacement != nil && !mp.txReplacement(oldScore.priority, priority) {
+			return fmt.Errorf("tx doesn't fit the replacement rule, oldPriority: %v, newPriority: %v", oldScore.priority, priority)
+		}
+
 		mp.priorityIndex.Remove(txMeta{
 			nonce:    nonce,
 			sender:   sender,
@@ -190,6 +197,12 @@ func (mp *priorityNonceMempool) Insert(ctx context.Context, tx sdk.Tx) error {
 		mp.priorityCounts[oldScore.priority]--
 	}
 
+	mp.priorityCounts[priority]++
+
+	// Since senderIndex is scored by nonce, a changed priority will overwrite the
+	// existing key.
+	key.senderElement = senderIndex.Set(key, tx)
+
 	mp.scores[sk] = txMeta{priority: priority}
 	mp.priorityIndex.Set(key, tx)
 

types/mempool/priority_nonce_test.go

@@ -638,3 +638,55 @@ func TestTxLimit(t *testing.T) {
 		require.Equal(t, 0, mp.CountTx())
 	}
 }
+
+func TestTxReplacement(t *testing.T) {
+	accounts := simtypes.RandomAccounts(rand.New(rand.NewSource(0)), 1)
+	ctx := sdk.NewContext(nil, tmproto.Header{}, false, log.NewNopLogger())
+	sa := accounts[0].Address
+
+	txs := []testTx{
+		{priority: 20, nonce: 1, address: sa},
+		{priority: 15, nonce: 1, address: sa}, // priority is less than the first Tx, failed tx replacement when the option enabled.
+		{priority: 23, nonce: 1, address: sa}, // priority is not 20% more than the first Tx, failed tx replacement when the option enabled.
+		{priority: 24, nonce: 1, address: sa}, // priority is 20% more than the first Tx, the first tx will be replaced.
+	}
+
+	// test Priority with default mempool
+	mp := mempool.NewPriorityMempool()
+	for _, tx := range txs {
+		c := ctx.WithPriority(tx.priority)
+		require.NoError(t, mp.Insert(c, tx))
+		require.Equal(t, 1, mp.CountTx())
+
+		iter := mp.Select(ctx, nil)
+		require.Equal(t, tx, iter.Tx())
+	}
+
+	// test Priority with TxReplacement
+	// we set a TestTxReplacement rule which the priority of the new Tx must be 20% more than the priority of the old Tx
+	// otherwise, the Insert will return error
+	feeBump := 20
+	mp = mempool.NewPriorityMempool(mempool.PriorityNonceWithTxReplacement(func(oldTxPriority, newTxPriority int64) bool {
+		threshold := int64(100 + feeBump)
+		return newTxPriority >= oldTxPriority*threshold/100
+	}))
+
+	c := ctx.WithPriority(txs[0].priority)
+	require.NoError(t, mp.Insert(c, txs[0]))
+	require.Equal(t, 1, mp.CountTx())
+
+	c = ctx.WithPriority(txs[1].priority)
+	require.Error(t, mp.Insert(c, txs[1]))
+	require.Equal(t, 1, mp.CountTx())
+
+	c = ctx.WithPriority(txs[2].priority)
+	require.Error(t, mp.Insert(c, txs[2]))
+	require.Equal(t, 1, mp.CountTx())
+
+	c = ctx.WithPriority(txs[3].priority)
+	require.NoError(t, mp.Insert(c, txs[3]))
+	require.Equal(t, 1, mp.CountTx())
+
+	iter := mp.Select(ctx, nil)
+	require.Equal(t, txs[3], iter.Tx())
+}