core: Limit total size of transaction pool

core/tx_list.go

@@ -466,6 +466,50 @@ func (l *txPricedList) Cap(threshold *big.Int, local *accountSet) types.Transact
 	return drop
 }
 
+// FreeSize drops transactions that sum up to at least a certain total size.
+func (l *txPricedList) FreeSize(sizeToFree common.StorageSize, threshold *big.Int, locals *accountSet) (types.Transactions, bool) {
+	drop := make(types.Transactions, 0, 128) // Remote underpriced transactions to drop
+	save := make(types.Transactions, 0, 64)  // Local underpriced transactions to keep
+	totalSize := common.StorageSize(0)
+
+	for len(*l.items) > 0 {
+		// Discard stale transactions if found during cleanup
+		tx := heap.Pop(l.items).(*types.Transaction)
+		if l.all.Get(tx.Hash()) == nil {
+			l.stales--
+			continue
+		}
+		// Stop the discards if we've reached the threshold
+		if tx.GasPrice().Cmp(threshold) >= 0 {
+			save = append(save, tx)
+			break
+		}
+		// Keep local
+		if locals.containsTx(tx) {
+			save = append(save, tx)
+			continue
+		}
+		// Non stale transaction found
+		drop = append(drop, tx)
+		totalSize += tx.Size()
+		// Stop the discards if we've reached the required size
+		if totalSize >= sizeToFree {
+			break
+		}
+	}
+	for _, tx := range save {
+		heap.Push(l.items, tx)
+	}
+	// If not enough space, bring back removed transactions
+	if totalSize < sizeToFree {
+		for _, tx := range drop {
+			heap.Push(l.items, tx)
+		}
+		return make(types.Transactions, 0), false
+	}
+	return drop, true
+}
+
 // Underpriced checks whether a transaction is cheaper than (or as cheap as) the
 // lowest priced transaction currently being tracked.
 func (l *txPricedList) Underpriced(tx *types.Transaction, local *accountSet) bool {

core/tx_pool.go

@@ -48,6 +48,10 @@ var (
 	// one present in the local chain.
 	ErrNonceTooLow = errors.New("nonce too low")
 
+	// ErrOverflown is returned if a transaction's size will overflow the allowed memory
+	// usage.
+	ErrOverflown = errors.New("transaction overflows memory")
+
 	// ErrUnderpriced is returned if a transaction's gas price is below the minimum
 	// configured for the transaction pool.
 	ErrUnderpriced = errors.New("transaction underpriced")
@@ -99,6 +103,7 @@ var (
 	// General tx metrics
 	validMeter         = metrics.NewRegisteredMeter("txpool/valid", nil)
 	invalidTxMeter     = metrics.NewRegisteredMeter("txpool/invalid", nil)
+	overflownTxMeter   = metrics.NewRegisteredMeter("txpool/overflown", nil)
 	underpricedTxMeter = metrics.NewRegisteredMeter("txpool/underpriced", nil)
 
 	pendingCounter = metrics.NewRegisteredCounter("txpool/pending", nil)
@@ -141,6 +146,8 @@ type TxPoolConfig struct {
 	AccountQueue uint64 // Maximum number of non-executable transaction slots permitted per account
 	GlobalQueue  uint64 // Maximum number of non-executable transaction slots for all accounts
 
+	MemoryLimit common.StorageSize // Limit on the total size of transactions in the pool.
+
 	Lifetime time.Duration // Maximum amount of time non-executable transaction are queued
 }
 
@@ -158,6 +165,9 @@ var DefaultTxPoolConfig = TxPoolConfig{
 	AccountQueue: 64,
 	GlobalQueue:  1024,
 
+	// Default 160MB total tx size limit
+	MemoryLimit: common.StorageSize(160 * 1024 * 1024),
+
 	Lifetime: 3 * time.Hour,
 }
 
@@ -507,8 +517,8 @@ func (pool *TxPool) local() map[common.Address]types.Transactions {
 // validateTx checks whether a transaction is valid according to the consensus
 // rules and adheres to some heuristic limits of the local node (price and size).
 func (pool *TxPool) validateTx(tx *types.Transaction, local bool) error {
-	// Heuristic limit, reject transactions over 32KB to prevent DOS attacks
+	// Heuristic limit, reject transactions over 120KB to prevent DOS attacks
-	if tx.Size() > 32*1024 {
+	if tx.Size() > 120*1024 {
 		return ErrOversizedData
 	}
 	// Transactions can't be negative. This may never happen using RLP decoded
@@ -588,6 +598,32 @@ func (pool *TxPool) add(tx *types.Transaction, local bool) (replaced bool, err e
 			pool.removeTx(tx.Hash(), false)
 		}
 	}
+	// If there isn't enough free size in pool, discard underpriced transactions
+	if pool.all.memory+tx.Size() > pool.config.MemoryLimit {
+		sizeToFree := pool.all.memory + tx.Size() - pool.config.MemoryLimit
+		gasPrice := tx.GasPrice()
+		if local {
+			// TODO: infinity.
+			gasPrice = big.NewInt(1000000000)
+		}
+		drop, success := pool.priced.FreeSize(sizeToFree, gasPrice, pool.locals)
+		if !local && !success {
+			log.Trace("Discarding underpriced transaction", "hash", hash, "price", tx.GasPrice())
+			overflownTxMeter.Mark(1)
+
+			// Bring back freed transactions
+			for _, tx := range drop {
+				pool.priced.Put(tx)
+			}
+			return false, ErrOverflown
+		}
+		// New transaction is better than our worse ones, make room for it
+		for _, tx := range drop {
+			log.Trace("Discarding freshly underpriced transaction", "hash", tx.Hash(), "price", tx.GasPrice())
+			overflownTxMeter.Mark(1)
+			pool.removeTx(tx.Hash(), false)
+		}
+	}
 
 	// Try to replace an existing transaction in the pending pool
 	from, _ := types.Sender(pool.signer, tx) // already validated
@@ -1462,6 +1498,7 @@ func (as *accountSet) merge(other *accountSet) {
 // TxPool.mu mutex.
 type txLookup struct {
 	all    map[common.Hash]*types.Transaction
+	memory common.StorageSize
 	lock   sync.RWMutex
 }
 
@@ -1505,6 +1542,7 @@ func (t *txLookup) Add(tx *types.Transaction) {
 	t.lock.Lock()
 	defer t.lock.Unlock()
 
+	t.memory += tx.Size()
 	t.all[tx.Hash()] = tx
 }
 
@@ -1513,5 +1551,6 @@ func (t *txLookup) Remove(hash common.Hash) {
 	t.lock.Lock()
 	defer t.lock.Unlock()
 
+	t.memory -= t.all[hash].Size()
 	delete(t.all, hash)
 }

core/tx_pool_test.go

@@ -77,6 +77,13 @@ func pricedTransaction(nonce uint64, gaslimit uint64, gasprice *big.Int, key *ec
 	return tx
 }
 
+func pricedDataTransaction(nonce uint64, gaslimit uint64, gasprice *big.Int, key *ecdsa.PrivateKey, numDataBytes uint64) *types.Transaction {
+	data := make([]byte, numDataBytes)
+	rand.Read(data)
+	tx, _ := types.SignTx(types.NewTransaction(nonce, common.Address{}, big.NewInt(100), gaslimit, gasprice, data), types.HomesteadSigner{}, key)
+	return tx
+}
+
 func setupTxPool() (*TxPool, *ecdsa.PrivateKey) {
 	statedb, _ := state.New(common.Hash{}, state.NewDatabase(rawdb.NewMemoryDatabase()))
 	blockchain := &testBlockChain{statedb, 1000000, new(event.Feed)}
@@ -113,6 +120,15 @@ func validateTxPoolInternals(pool *TxPool) error {
 			return fmt.Errorf("pending nonce mismatch: have %v, want %v", nonce, last+1)
 		}
 	}
+	// Ensure the total size of the transactions match the running counter
+	memory := common.StorageSize(0)
+	for _, tx := range pool.all.all {
+		memory += tx.Size()
+	}
+	if pool.all.memory != memory {
+		return fmt.Errorf("tracked total transaction size %d != calculated total size %d",
+			uint64(pool.all.memory), uint64(memory))
+	}
 	return nil
 }
 
@@ -1002,6 +1018,59 @@ func TestTransactionPendingGlobalLimiting(t *testing.T) {
 	}
 }
 
+// Tests that if the total transaction size in the pool goes above some
+// hard threshold, the higher transactions are dropped to prevent DOS attacks.
+func TestTransactionTotalSizeLimiting(t *testing.T) {
+	t.Parallel()
+
+	// Create the pool to test the limit enforcement with
+	statedb, _ := state.New(common.Hash{}, state.NewDatabase(rawdb.NewMemoryDatabase()))
+	blockchain := &testBlockChain{statedb, 1000000, new(event.Feed)}
+
+	config := testTxPoolConfig
+	config.MemoryLimit = common.StorageSize(10 * 1000)
+
+	pool := NewTxPool(config, params.TestChainConfig, blockchain)
+	defer pool.Stop()
+
+	// Create a number of test accounts and fund them
+	keys := make([]*ecdsa.PrivateKey, 5)
+	for i := 0; i < len(keys); i++ {
+		keys[i], _ = crypto.GenerateKey()
+		pool.currentState.AddBalance(crypto.PubkeyToAddress(keys[i].PublicKey), big.NewInt(1000000))
+	}
+	// Generate and queue a batch of transactions
+	nonces := make(map[common.Address]uint64)
+
+	txs := types.Transactions{}
+	for _, key := range keys {
+		addr := crypto.PubkeyToAddress(key.PublicKey)
+		for j := 0; j < int(config.GlobalSlots)/len(keys)*2; j++ {
+			datalen := uint64(rand.Intn(1000))
+			price := big.NewInt(int64(rand.Intn(10)))
+			tx := pricedDataTransaction(nonces[addr], 100000, price, key, datalen)
+			txs = append(txs, tx)
+			nonces[addr]++
+		}
+	}
+	// Import the batch and verify that limits have been enforced
+	pool.AddRemotes(txs)
+
+	if pool.all.memory > config.MemoryLimit {
+		t.Fatalf("Pool size too large: %d. Expected at most %d", uint64(pool.all.memory), uint64(config.MemoryLimit))
+	}
+	if err := validateTxPoolInternals(pool); err != nil {
+		t.Fatalf("pool internal state corrupted: %v", err)
+	}
+
+	for _, tx := range pool.all.all {
+		// Remaining transactions should all have high price. The highest in our range is 9.
+		if tx.GasPrice().Int64() < 9 {
+			t.Fatalf("Remaining transaction with low price")
+		}
+	}
+}
+
 // Tests that if transactions start being capped, transactions are also removed from 'all'
 func TestTransactionCapClearsFromAll(t *testing.T) {
 	t.Parallel()