core/txpool: implement account exclusivity across subpools

core/txpool/blobpool/blobpool.go

@@ -288,7 +288,8 @@ type BlobTxShim struct {
 //     minimums will need to be done only starting at the swapped in/out nonce
 //     and leading up to the first no-change.
 type BlobPool struct {
-	config Config // Pool configuration
+	config  Config                 // Pool configuration
+	reserve txpool.AddressReserver // Address reserver to ensure exclusivity across subpools
 
 	store  billy.Database // Persistent data store for the tx metadata and blobs
 	stored uint64         // Useful data size of all transactions on disk
@@ -337,7 +338,9 @@ func (p *BlobPool) Filter(tx *types.Transaction) bool {
 // Init sets the gas price needed to keep a transaction in the pool and the chain
 // head to allow balance / nonce checks. The transaction journal will be loaded
 // from disk and filtered based on the provided starting settings.
-func (p *BlobPool) Init(gasTip *big.Int, head *types.Header) error {
+func (p *BlobPool) Init(gasTip *big.Int, head *types.Header, reserve txpool.AddressReserver) error {
+	p.reserve = reserve
+
 	var (
 		queuedir = filepath.Join(p.config.Datadir, pendingTransactionStore)
 		limbodir = filepath.Join(p.config.Datadir, limboedTransactionStore)
@@ -454,6 +457,9 @@ func (p *BlobPool) parseTransaction(id uint64, size uint32, blob []byte) error {
 		return err
 	}
 	if _, ok := p.index[sender]; !ok {
+		if err := p.reserve(sender, true); err != nil {
+			return err
+		}
 		p.index[sender] = []*blobTxMeta{}
 		p.spent[sender] = new(uint256.Int)
 	}
@@ -507,6 +513,8 @@ func (p *BlobPool) recheck(addr common.Address, inclusions map[common.Hash]uint6
 		if inclusions != nil { // only during reorgs will the heap will be initialized
 			heap.Remove(p.evict, p.evict.index[addr])
 		}
+		p.reserve(addr, false)
+
 		if gapped {
 			log.Warn("Dropping dangling blob transactions", "from", addr, "missing", next, "drop", nonces, "ids", ids)
 		} else {
@@ -631,6 +639,7 @@ func (p *BlobPool) recheck(addr common.Address, inclusions map[common.Hash]uint6
 			if inclusions != nil { // only during reorgs will the heap will be initialized
 				heap.Remove(p.evict, p.evict.index[addr])
 			}
+			p.reserve(addr, false)
 		} else {
 			p.index[addr] = txs
 		}
@@ -868,6 +877,10 @@ func (p *BlobPool) reinject(addr common.Address, tx *types.Transaction) {
 	meta := newBlobTxMeta(id, p.store.Size(id), tx)
 
 	if _, ok := p.index[addr]; !ok {
+		if err := p.reserve(addr, true); err != nil {
+			log.Warn("Failed to reserve account for blob pool", "tx", tx.Hash(), "from", addr, "err", err)
+			return
+		}
 		p.index[addr] = []*blobTxMeta{meta}
 		p.spent[addr] = meta.costCap
 		p.evict.Push(addr)
@@ -920,6 +933,7 @@ func (p *BlobPool) SetGasTip(tip *big.Int) {
 					} else {
 						delete(p.index, addr)
 						delete(p.spent, addr)
+						p.reserve(addr, false)
 					}
 					// Clear out the transactions from the data store
 					log.Warn("Dropping underpriced blob transaction", "from", addr, "rejected", tx.nonce, "tip", tx.execTipCap, "want", tip, "drop", nonces, "ids", ids)
@@ -1064,7 +1078,7 @@ func (p *BlobPool) Add(txs []*txpool.Transaction, local bool, sync bool) []error
 
 // Add inserts a new blob transaction into the pool if it passes validation (both
 // consensus validity and pool restictions).
-func (p *BlobPool) add(tx *types.Transaction, blobs []kzg4844.Blob, commits []kzg4844.Commitment, proofs []kzg4844.Proof) error {
+func (p *BlobPool) add(tx *types.Transaction, blobs []kzg4844.Blob, commits []kzg4844.Commitment, proofs []kzg4844.Proof) (err error) {
 	// The blob pool blocks on adding a transaction. This is because blob txs are
 	// only even pulled form the network, so this method will act as the overload
 	// protection for fetches.
@@ -1082,6 +1096,25 @@ func (p *BlobPool) add(tx *types.Transaction, blobs []kzg4844.Blob, commits []kz
 		log.Trace("Transaction validation failed", "hash", tx.Hash(), "err", err)
 		return err
 	}
+	// If the address is not yet known, request exclusivity to track the account
+	// only by this subpool until all transactions are evicted
+	from, _ := types.Sender(p.signer, tx) // already validated above
+	if _, ok := p.index[from]; !ok {
+		if err := p.reserve(from, true); err != nil {
+			return err
+		}
+		defer func() {
+			// If the transaction is rejected by some post-validation check, remove
+			// the lock on the reservation set.
+			//
+			// Note, `err` here is the named error return, which will be initialized
+			// by a return statement before running deferred methods. Take care with
+			// removing or subscoping err as it will break this clause.
+			if err != nil {
+				p.reserve(from, false)
+			}
+		}()
+	}
 	// Transaction permitted into the pool from a nonce and cost perspective,
 	// insert it into the database and update the indices
 	blob, err := rlp.EncodeToBytes(&blobTx{Tx: tx, Blobs: blobs, Commits: commits, Proofs: proofs})
@@ -1096,10 +1129,9 @@ func (p *BlobPool) add(tx *types.Transaction, blobs []kzg4844.Blob, commits []kz
 	meta := newBlobTxMeta(id, p.store.Size(id), tx)
 
 	var (
-		from, _ = types.Sender(p.signer, tx) // already validated above
-		next    = p.state.GetNonce(from)
-		offset  = int(tx.Nonce() - next)
-		newacc  = false
+		next   = p.state.GetNonce(from)
+		offset = int(tx.Nonce() - next)
+		newacc = false
 	)
 	if len(p.index[from]) > offset {
 		// Transaction replaces a previously queued one
@@ -1217,6 +1249,7 @@ func (p *BlobPool) drop() {
 	if last {
 		delete(p.index, from)
 		delete(p.spent, from)
+		p.reserve(from, false)
 	} else {
 		txs[len(txs)-1] = nil
 		txs = txs[:len(txs)-1]

core/txpool/blobpool/blobpool_test.go

@@ -24,6 +24,7 @@ import (
 	"math/big"
 	"os"
 	"path/filepath"
+	"sync"
 	"testing"
 	"time"
 
@@ -159,6 +160,33 @@ func (bc *testBlockChain) StateAt(common.Hash) (*state.StateDB, error) {
 	return bc.statedb, nil
 }
 
+// makeAddressReserver is a utility method to sanity check that accounts are
+// properly reserved by the blobpool (no duplicate reserves or unreserves).
+func makeAddressReserver() txpool.AddressReserver {
+	var (
+		reserved = make(map[common.Address]struct{})
+		lock     sync.Mutex
+	)
+	return func(addr common.Address, reserve bool) error {
+		lock.Lock()
+		defer lock.Unlock()
+
+		_, exists := reserved[addr]
+		if reserve {
+			if exists {
+				panic("already reserved")
+			}
+			reserved[addr] = struct{}{}
+			return nil
+		}
+		if !exists {
+			panic("not reserved")
+		}
+		delete(reserved, addr)
+		return nil
+	}
+}
+
 // makeTx is a utility method to construct a random blob transaction and sign it
 // with a valid key, only setting the interesting fields from the perspective of
 // the blob pool.
@@ -460,7 +488,7 @@ func TestOpenDrops(t *testing.T) {
 		statedb: statedb,
 	}
 	pool := New(Config{Datadir: storage}, chain)
-	if err := pool.Init(big.NewInt(1), chain.CurrentBlock()); err != nil {
+	if err := pool.Init(big.NewInt(1), chain.CurrentBlock(), makeAddressReserver()); err != nil {
 		t.Fatalf("failed to create blob pool: %v", err)
 	}
 	defer pool.Close()
@@ -569,7 +597,7 @@ func TestOpenIndex(t *testing.T) {
 		statedb: statedb,
 	}
 	pool := New(Config{Datadir: storage}, chain)
-	if err := pool.Init(big.NewInt(1), chain.CurrentBlock()); err != nil {
+	if err := pool.Init(big.NewInt(1), chain.CurrentBlock(), makeAddressReserver()); err != nil {
 		t.Fatalf("failed to create blob pool: %v", err)
 	}
 	defer pool.Close()
@@ -660,7 +688,7 @@ func TestOpenHeap(t *testing.T) {
 		statedb: statedb,
 	}
 	pool := New(Config{Datadir: storage}, chain)
-	if err := pool.Init(big.NewInt(1), chain.CurrentBlock()); err != nil {
+	if err := pool.Init(big.NewInt(1), chain.CurrentBlock(), makeAddressReserver()); err != nil {
 		t.Fatalf("failed to create blob pool: %v", err)
 	}
 	defer pool.Close()
@@ -740,7 +768,7 @@ func TestOpenCap(t *testing.T) {
 			statedb: statedb,
 		}
 		pool := New(Config{Datadir: storage, Datacap: datacap}, chain)
-		if err := pool.Init(big.NewInt(1), chain.CurrentBlock()); err != nil {
+		if err := pool.Init(big.NewInt(1), chain.CurrentBlock(), makeAddressReserver()); err != nil {
 			t.Fatalf("failed to create blob pool: %v", err)
 		}
 		// Verify that enough transactions have been dropped to get the pool's size
@@ -1045,7 +1073,7 @@ func TestAdd(t *testing.T) {
 			statedb: statedb,
 		}
 		pool := New(Config{Datadir: storage}, chain)
-		if err := pool.Init(big.NewInt(1), chain.CurrentBlock()); err != nil {
+		if err := pool.Init(big.NewInt(1), chain.CurrentBlock(), makeAddressReserver()); err != nil {
 			t.Fatalf("test %d: failed to create blob pool: %v", i, err)
 		}
 		verifyPoolInternals(t, pool)

core/txpool/legacypool/legacypool.go

@@ -219,6 +219,7 @@ type LegacyPool struct {
 	locals  *accountSet // Set of local transaction to exempt from eviction rules
 	journal *journal    // Journal of local transaction to back up to disk
 
+	reserve txpool.AddressReserver       // Address reserver to ensure exclusivity across subpools
 	pending map[common.Address]*list     // All currently processable transactions
 	queue   map[common.Address]*list     // Queued but non-processable transactions
 	beats   map[common.Address]time.Time // Last heartbeat from each known account
@@ -291,7 +292,10 @@ func (pool *LegacyPool) Filter(tx *types.Transaction) bool {
 // head to allow balance / nonce checks. The transaction journal will be loaded
 // from disk and filtered based on the provided starting settings. The internal
 // goroutines will be spun up and the pool deemed operational afterwards.
-func (pool *LegacyPool) Init(gasTip *big.Int, head *types.Header) error {
+func (pool *LegacyPool) Init(gasTip *big.Int, head *types.Header, reserve txpool.AddressReserver) error {
+	// Set the address reserver to request exclusive access to pooled accounts
+	pool.reserve = reserve
+
 	// Set the basic pool parameters
 	pool.gasTip.Store(gasTip)
 	pool.reset(nil, head)
@@ -365,7 +369,7 @@ func (pool *LegacyPool) loop() {
 				if time.Since(pool.beats[addr]) > pool.config.Lifetime {
 					list := pool.queue[addr].Flatten()
 					for _, tx := range list {
-						pool.removeTx(tx.Hash(), true)
+						pool.removeTx(tx.Hash(), true, true)
 					}
 					queuedEvictionMeter.Mark(int64(len(list)))
 				}
@@ -428,7 +432,7 @@ func (pool *LegacyPool) SetGasTip(tip *big.Int) {
 		// pool.priced is sorted by GasFeeCap, so we have to iterate through pool.all instead
 		drop := pool.all.RemotesBelowTip(tip)
 		for _, tx := range drop {
-			pool.removeTx(tx.Hash(), false)
+			pool.removeTx(tx.Hash(), false, true)
 		}
 		pool.priced.Removed(len(drop))
 	}
@@ -643,10 +647,31 @@ func (pool *LegacyPool) add(tx *types.Transaction, local bool) (replaced bool, e
 		invalidTxMeter.Mark(1)
 		return false, err
 	}
-
 	// already validated by this point
 	from, _ := types.Sender(pool.signer, tx)
 
+	// If the address is not yet known, request exclusivity to track the account
+	// only by this subpool until all transactions are evicted
+	var (
+		_, hasPending = pool.pending[from]
+		_, hasQueued  = pool.queue[from]
+	)
+	if !hasPending && !hasQueued {
+		if err := pool.reserve(from, true); err != nil {
+			return false, err
+		}
+		defer func() {
+			// If the transaction is rejected by some post-validation check, remove
+			// the lock on the reservation set.
+			//
+			// Note, `err` here is the named error return, which will be initialized
+			// by a return statement before running deferred methods. Take care with
+			// removing or subscoping err as it will break this clause.
+			if err != nil {
+				pool.reserve(from, false)
+			}
+		}()
+	}
 	// If the transaction pool is full, discard underpriced transactions
 	if uint64(pool.all.Slots()+numSlots(tx)) > pool.config.GlobalSlots+pool.config.GlobalQueue {
 		// If the new transaction is underpriced, don't accept it
@@ -701,7 +726,10 @@ func (pool *LegacyPool) add(tx *types.Transaction, local bool) (replaced bool, e
 		for _, tx := range drop {
 			log.Trace("Discarding freshly underpriced transaction", "hash", tx.Hash(), "gasTipCap", tx.GasTipCap(), "gasFeeCap", tx.GasFeeCap())
 			underpricedTxMeter.Mark(1)
-			dropped := pool.removeTx(tx.Hash(), false)
+
+			sender, _ := types.Sender(pool.signer, tx)
+			dropped := pool.removeTx(tx.Hash(), false, sender != from) // Don't unreserve the sender of the tx being added if last from the acc
+
 			pool.changesSinceReorg += dropped
 		}
 	}
@@ -1025,15 +1053,35 @@ func (pool *LegacyPool) Has(hash common.Hash) bool {
 
 // removeTx removes a single transaction from the queue, moving all subsequent
 // transactions back to the future queue.
+//
+// In unreserve is false, the account will not be relinquished to the main txpool
+// even if there are no more references to it. This is used to handle a race when
+// a tx being added, and it evicts a previously scheduled tx from the same account,
+// which could lead to a premature release of the lock.
+//
 // Returns the number of transactions removed from the pending queue.
-func (pool *LegacyPool) removeTx(hash common.Hash, outofbound bool) int {
+func (pool *LegacyPool) removeTx(hash common.Hash, outofbound bool, unreserve bool) int {
 	// Fetch the transaction we wish to delete
 	tx := pool.all.Get(hash)
 	if tx == nil {
 		return 0
 	}
 	addr, _ := types.Sender(pool.signer, tx) // already validated during insertion
 
+	// If after deletion there are no more transactions belonging to this account,
+	// relinquish the address reservation. It's a bit convoluted do this, via a
+	// defer, but it's safer vs. the many return pathways.
+	if unreserve {
+		defer func() {
+			var (
+				_, hasPending = pool.pending[addr]
+				_, hasQueued  = pool.queue[addr]
+			)
+			if !hasPending && !hasQueued {
+				pool.reserve(addr, false)
+			}
+		}()
+	}
 	// Remove it from the list of known transactions
 	pool.all.Remove(hash)
 	if outofbound {
@@ -1419,6 +1467,9 @@ func (pool *LegacyPool) promoteExecutables(accounts []common.Address) []*types.T
 		if list.Empty() {
 			delete(pool.queue, addr)
 			delete(pool.beats, addr)
+			if _, ok := pool.pending[addr]; !ok {
+				pool.reserve(addr, false)
+			}
 		}
 	}
 	return promoted
@@ -1540,7 +1591,7 @@ func (pool *LegacyPool) truncateQueue() {
 		// Drop all transactions if they are less than the overflow
 		if size := uint64(list.Len()); size <= drop {
 			for _, tx := range list.Flatten() {
-				pool.removeTx(tx.Hash(), true)
+				pool.removeTx(tx.Hash(), true, true)
 			}
 			drop -= size
 			queuedRateLimitMeter.Mark(int64(size))
@@ -1549,7 +1600,7 @@ func (pool *LegacyPool) truncateQueue() {
 		// Otherwise drop only last few transactions
 		txs := list.Flatten()
 		for i := len(txs) - 1; i >= 0 && drop > 0; i-- {
-			pool.removeTx(txs[i].Hash(), true)
+			pool.removeTx(txs[i].Hash(), true, true)
 			drop--
 			queuedRateLimitMeter.Mark(1)
 		}
@@ -1611,6 +1662,9 @@ func (pool *LegacyPool) demoteUnexecutables() {
 		// Delete the entire pending entry if it became empty.
 		if list.Empty() {
 			delete(pool.pending, addr)
+			if _, ok := pool.queue[addr]; !ok {
+				pool.reserve(addr, false)
+			}
 		}
 	}
 }