fix race condition between RemoveTx and runReorg loop

make public RemoveTx be thread safe by acquiring the pool lock and
private removeTx does not

Author: Matt Acciai

CHANGELOG.md

@@ -16,6 +16,7 @@
 - [\#591](https://github.com/cosmos/evm/pull/591) CheckTxHandler should handle "invalid nonce" tx
 - [\#643](https://github.com/cosmos/evm/pull/643) Support for mnemonic source (file, stdin,etc) flag in key add command.
 - [\#645](https://github.com/cosmos/evm/pull/645) Align precise bank keeper for correct decimal conversion in evmd.
+* [\#658](https://github.com/cosmos/evm/pull/658) Fix race condition between legacypool's RemoveTx and runReorg.
 
 ### IMPROVEMENTS
 

mempool/txpool/legacypool/legacypool.go

@@ -377,7 +377,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, true)
+						pool.removeTx(tx.Hash(), true, true)
 					}
 					queuedEvictionMeter.Mark(int64(len(list)))
 				}
@@ -430,7 +430,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.TxsBelowTip(tip)
 		for _, tx := range drop {
-			pool.RemoveTx(tx.Hash(), false, true)
+			pool.removeTx(tx.Hash(), false, true)
 		}
 		pool.priced.Removed(len(drop))
 	}
@@ -770,7 +770,7 @@ func (pool *LegacyPool) add(tx *types.Transaction) (replaced bool, err error) {
 			underpricedTxMeter.Mark(1)
 
 			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
+			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
 		}
@@ -1087,6 +1087,23 @@ func (pool *LegacyPool) Has(hash common.Hash) bool {
 //
 // Returns the number of transactions removed from the pending queue.
 func (pool *LegacyPool) RemoveTx(hash common.Hash, outofbound bool, unreserve bool) int {
+	pool.mu.Lock()
+	defer pool.mu.Unlock()
+	return pool.removeTx(hash, outofbound, unreserve)
+}
+
+// removeTx removes a single transaction from the queue, moving all subsequent
+// transactions back to the future queue.
+//
+// If 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.
+//
+// The transaction pool lock must be held.
+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 {
@@ -1533,7 +1550,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, true)
+				pool.removeTx(tx.Hash(), true, true)
 			}
 			drop -= size
 			queuedRateLimitMeter.Mark(int64(size))
@@ -1542,7 +1559,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, true)
+			pool.removeTx(txs[i].Hash(), true, true)
 			drop--
 			queuedRateLimitMeter.Mark(1)
 		}

mempool/txpool/legacypool/legacypool_test.go

@@ -543,7 +543,7 @@ func TestChainFork(t *testing.T) {
 	if _, err := pool.add(tx); err != nil {
 		t.Error("didn't expect error", err)
 	}
-	pool.RemoveTx(tx.Hash(), true, true)
+	pool.removeTx(tx.Hash(), true, true)
 
 	// reset the pool's internal state
 	resetState()
@@ -2591,6 +2591,61 @@ func TestSetCodeTransactionsReorg(t *testing.T) {
 	}
 }
 
+// TestRemoveTxTruncatePoolRace is a regression test for a race condition
+// between removing txs and runReorg loop. Run this with the -race flag to
+// ensure that there is no race condition between the two functions.
+func TestRemoveTxTruncatePoolRace(t *testing.T) {
+	statedb, _ := state.New(types.EmptyRootHash, state.NewDatabaseForTesting())
+	blockchain := newTestBlockChain(params.MergedTestChainConfig, 1000000, statedb, new(event.Feed))
+
+	pool := New(testTxPoolConfig, blockchain)
+	err := pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), newReserver())
+	if err != nil {
+		t.Fatalf("failed to init pool: %v", err)
+	}
+
+	// fill the pool with txs
+	fillPool(t, pool)
+
+	// make a copy of all hashes in the pool so that we do not have to iterate
+	// over pending and queue while we call RemoveTx, potentially triggering
+	// the race condition ourselves
+	var hashes []common.Hash
+	for _, txs := range pool.pending {
+		for _, tx := range txs.Flatten() {
+			hashes = append(hashes, tx.Hash())
+		}
+	}
+	for _, txs := range pool.queue {
+		for _, tx := range txs.Flatten() {
+			hashes = append(hashes, tx.Hash())
+		}
+	}
+
+	var wg sync.WaitGroup
+
+	// manually trigger the reorg loop to run (5 times just to ensure that we
+	// will trigger the race condition)
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for range 5 {
+			pool.runReorg(make(chan struct{}), nil, nil, nil)
+		}
+	}()
+
+	// call RemoveTx on every tx in the pool
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for _, hash := range hashes {
+			_ = pool.RemoveTx(hash, false, true)
+		}
+	}()
+
+	wg.Wait()
+}
+
 // Benchmarks the speed of validating the contents of the pending queue of the
 // transaction pool.
 func BenchmarkPendingDemotion100(b *testing.B)   { benchmarkPendingDemotion(b, 100) }