Drain replacement fix

txpool/pool.go

@@ -482,8 +482,12 @@ func (p *TxPool) processRemoteTxs(ctx context.Context) error {
 		return err
 	}
 
+	add := func(mt *metaTx, announcements *types.Announcements) DiscardReason {
+		return p.addLockedWithChecks(mt, announcements, cacheView)
+	}
 	announcements, _, err := addTxs(p.lastSeenBlock.Load(), cacheView, p.senders, newTxs,
-		p.pendingBaseFee.Load(), p.blockGasLimit.Load(), p.pending, p.baseFee, p.queued, p.all, p.byHash, p.addLocked, p.discardLocked, true)
+		p.pendingBaseFee.Load(), p.blockGasLimit.Load(), p.pending, p.baseFee, p.queued, p.all, p.byHash, add, p.discardLocked, true)
+
 	if err != nil {
 		return err
 	}
@@ -1074,6 +1078,41 @@ func (p *TxPool) setBaseFee(baseFee uint64) (uint64, bool) {
 	return p.pendingBaseFee.Load(), changed
 }
 
+func (p *TxPool) addLockedWithChecks(mt *metaTx, announcements *types.Announcements, cacheView kvcache.CacheView) DiscardReason {
+	// Discard the transaction if it would invalidate others in the pool by overdraft
+	senderID := mt.Tx.SenderID
+	if p.all.count(senderID) >= 2 {
+		nonce, currentBalance, err := p.senders.info(cacheView, senderID)
+		if err != nil {
+			log.Error("Could not fetch sender's balance:", "err", err)
+			return NotReplaced // TODO better fitting code?
+		}
+		overdraft := false
+		balance := &currentBalance
+
+		p.all.ascendFrom(nonce, senderID, func(mt1 *metaTx) bool {
+			mt2 := mt1
+			if mt1.Tx.Nonce == mt.Tx.Nonce {
+				mt2 = mt
+			}
+
+			needBalance := uint256.NewInt(mt2.Tx.Gas)
+			needBalance.Mul(needBalance, &mt2.Tx.FeeCap)
+			needBalance.Add(needBalance, &mt.Tx.Value)
+
+			_, overdraft = balance.SubOverflow(balance, needBalance)
+
+			return !overdraft
+		})
+
+		if overdraft {
+			return InsufficientFunds
+		}
+	}
+
+	return p.addLocked(mt, announcements)
+}
+
 func (p *TxPool) addLocked(mt *metaTx, announcements *types.Announcements) DiscardReason {
 	// Insert to pending pool, if pool doesn't have txn with same Nonce and bigger Tip
 	found := p.all.get(mt.Tx.SenderID, mt.Tx.Nonce)
@@ -1199,7 +1238,7 @@ func removeMined(byNonce *BySenderAndNonce, minedTxs []*types.TxSlot, pending *P
 }
 
 // onSenderStateChange is the function that recalculates ephemeral fields of transactions and determines
-// which sub pool they will need to go to. Sice this depends on other transactions from the same sender by with lower
+// which sub pool they will need to go to. Since this depends on other transactions from the same sender by with lower
 // nonces, and also affect other transactions from the same sender with higher nonce, it loops through all transactions
 // for a given senderID
 func onSenderStateChange(senderID uint64, senderNonce uint64, senderBalance uint256.Int, byNonce *BySenderAndNonce,
@@ -1694,6 +1733,7 @@ func (p *TxPool) fromDB(ctx context.Context, tx kv.Tx, coreTx kv.Tx) error {
 	if err != nil {
 		return err
 	}
+
 	if _, _, err := addTxs(p.lastSeenBlock.Load(), cacheView, p.senders, txs,
 		pendingBaseFee, math.MaxUint64 /* blockGasLimit */, p.pending, p.baseFee, p.queued, p.all, p.byHash, p.addLocked, p.discardLocked, false); err != nil {
 		return err
@@ -2040,17 +2080,20 @@ func (b *BySenderAndNonce) ascendAll(f func(*metaTx) bool) {
 		return f(mt)
 	})
 }
-func (b *BySenderAndNonce) ascend(senderID uint64, f func(*metaTx) bool) {
+func (b *BySenderAndNonce) ascendFrom(nonce uint64, senderID uint64, f func(*metaTx) bool) {
 	s := b.search
 	s.Tx.SenderID = senderID
-	s.Tx.Nonce = 0
+	s.Tx.Nonce = nonce
 	b.tree.AscendGreaterOrEqual(s, func(mt *metaTx) bool {
 		if mt.Tx.SenderID != senderID {
 			return false
 		}
 		return f(mt)
 	})
 }
+func (b *BySenderAndNonce) ascend(senderID uint64, f func(*metaTx) bool) {
+	b.ascendFrom(0, senderID, f)
+}
 func (b *BySenderAndNonce) descend(senderID uint64, f func(*metaTx) bool) {
 	s := b.search
 	s.Tx.SenderID = senderID

txpool/pool_test.go

@@ -712,3 +712,128 @@ func TestShanghaiValidateTx(t *testing.T) {
 		})
 	}
 }
+
+func TestOverdraftByReplacement(t *testing.T) {
+	assert, require := assert.New(t), require.New(t)
+	ch := make(chan types.Announcements, 100)
+	db, coreDB := memdb.NewTestPoolDB(t), memdb.NewTestDB(t)
+
+	cfg := txpoolcfg.DefaultConfig
+	cfg.PendingSubPoolLimit = 30
+	cfg.AccountSlots = 6
+
+	sendersCache := kvcache.New(kvcache.DefaultCoherentConfig)
+	pool, err := New(ch, coreDB, cfg, sendersCache, *u256.N1, nil)
+	assert.NoError(err)
+	require.True(pool != nil)
+
+	ctx := context.Background()
+	var stateVersionID uint64 = 0
+	pendingBaseFee := uint64(200000)
+
+	h1 := gointerfaces.ConvertHashToH256([32]byte{})
+	changes := &remote.StateChangeBatch{
+		StateVersionId:      stateVersionID,
+		PendingBlockBaseFee: pendingBaseFee,
+		BlockGasLimit:       1_000_000,
+		ChangeBatch: []*remote.StateChange{
+			{BlockHeight: 0, BlockHash: h1},
+		},
+	}
+
+	var addrs [10][20]byte
+	balance := uint256.NewInt(1 * common.Ether)
+	for i := 0; i < 10; i++ {
+		addrs[i][0] = byte(i + 1)
+		v := make([]byte, types.EncodeSenderLengthForStorage(1, *balance))
+		types.EncodeSender(1, *balance, v)
+		changes.ChangeBatch[0].Changes = append(changes.ChangeBatch[0].Changes, &remote.AccountChange{
+			Action:  remote.Action_UPSERT,
+			Address: gointerfaces.ConvertAddressToH160(addrs[i]),
+			Data:    v,
+		})
+	}
+
+	tx, err := db.BeginRw(ctx)
+	require.NoError(err)
+	defer tx.Rollback()
+	err = pool.OnNewBlock(ctx, changes, types.TxSlots{}, types.TxSlots{}, tx)
+	assert.NoError(err)
+
+	var txSlots1 types.TxSlots
+
+	// legit transactions
+	for i := 0; i < 5; i++ {
+		txSlot := &types.TxSlot{
+			Tip:    *uint256.NewInt(300000),
+			FeeCap: *uint256.NewInt(300000),
+			Gas:    100000,
+			Nonce:  1,
+		}
+		txSlot.IDHash[0] = byte(i + 1)
+		txSlots1.Append(txSlot, addrs[i][:], true)
+	}
+
+	pool.AddRemoteTxs(ctx, txSlots1)
+	err = pool.processRemoteTxs(ctx)
+	assert.NoError(err)
+
+	// replacement transactions to fill the pool
+	var txSlots2 types.TxSlots
+	replacementTxValue := uint256.NewInt(common.Ether / 10)
+	for i := 5; i < 10; i++ {
+		for j := 1; j <= 6; j++ {
+			txSlot := &types.TxSlot{
+				Tip:    *uint256.NewInt(400000),
+				Value:  *replacementTxValue,
+				FeeCap: *uint256.NewInt(400000),
+				Gas:    200000,
+				Nonce:  uint64(j),
+			}
+			txSlot.IDHash[0] = byte(i*10 + j)
+			txSlots2.Append(txSlot, addrs[i][:], true)
+		}
+	}
+
+	pool.AddRemoteTxs(ctx, txSlots2)
+	err = pool.processRemoteTxs(ctx)
+	assert.NoError(err)
+
+	// pending pool is full and the legit transactions are out
+	assert.Equal(30, pool.pending.Len())
+	for _, pendingTx := range pool.pending.best.ms {
+		senderID := (*pendingTx).Tx.SenderID
+		assert.LessOrEqual(senderID, uint64(10))
+		assert.GreaterOrEqual(senderID, uint64(6))
+	}
+
+	// draining transactions
+	var txSlots3 types.TxSlots
+	drainTxValue := uint256.NewInt(common.Ether * 0.95)
+	for i := 5; i < 10; i++ {
+		txSlot := &types.TxSlot{
+			Tip:    *uint256.NewInt(500000),
+			Value:  *drainTxValue,
+			FeeCap: *uint256.NewInt(500000),
+			Gas:    300000,
+			Nonce:  1,
+		}
+		txSlot.IDHash[0] = byte(i*10 + 9)
+		txSlots3.Append(txSlot, addrs[i][:], true)
+	}
+
+	pool.AddRemoteTxs(ctx, txSlots3)
+	err = pool.processRemoteTxs(ctx)
+	assert.NoError(err)
+
+	// expected behavior:
+	// have been rejected, so replacement tx are still there
+	assert.Equal(30, pool.pending.Len())
+	for _, pendingTx := range pool.pending.best.ms {
+		value := (*pendingTx).Tx.Value
+		assert.Equal(*replacementTxValue, value)
+		senderID := (*pendingTx).Tx.SenderID
+		assert.LessOrEqual(senderID, uint64(10))
+		assert.GreaterOrEqual(senderID, uint64(6))
+	}
+}