From 6baf9fdf0b743bd1b24962d482d5407877c9e359 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C3=96mer=20Faruk=20Irmak?= Date: Fri, 20 Sep 2024 15:46:09 +0300 Subject: [PATCH] feat(txpool): only try demoting txns from accounts that were active txpool demotes pending txns only if their nonce is now lower than the nonce in the latest state or the account no longer has enough funds to cover the costs. Unless the account in question was active since the last txpool reorg, there is no way that it's nonce changed or balance decreased. --- core/txpool/legacypool/legacypool.go | 43 ++++++++++++++++------- core/txpool/legacypool/legacypool_test.go | 2 +- 2 files changed, 32 insertions(+), 13 deletions(-) diff --git a/core/txpool/legacypool/legacypool.go b/core/txpool/legacypool/legacypool.go index 5506ecc31fc3..04ce48d5ba4f 100644 --- a/core/txpool/legacypool/legacypool.go +++ b/core/txpool/legacypool/legacypool.go @@ -1272,9 +1272,10 @@ func (pool *LegacyPool) runReorg(done chan struct{}, reset *txpoolResetRequest, promoteAddrs = dirtyAccounts.flatten() } pool.mu.Lock() + var demoteAddrs map[common.Address]bool if reset != nil { // Reset from the old head to the new, rescheduling any reorged transactions - pool.reset(reset.oldHead, reset.newHead) + demoteAddrs = pool.reset(reset.oldHead, reset.newHead) // Nonces were reset, discard any events that became stale for addr := range events { @@ -1296,7 +1297,7 @@ func (pool *LegacyPool) runReorg(done chan struct{}, reset *txpoolResetRequest, // remove any transaction that has been included in the block or was invalidated // because of another transaction (e.g. higher gas price). if reset != nil { - pool.demoteUnexecutables() + pool.demoteUnexecutables(demoteAddrs) if reset.newHead != nil { if pool.chainconfig.IsLondon(new(big.Int).Add(reset.newHead.Number, big.NewInt(1))) { pendingBaseFee := eip1559.CalcBaseFee(pool.chainconfig, reset.newHead) @@ -1340,10 +1341,18 @@ func (pool *LegacyPool) runReorg(done chan struct{}, reset *txpoolResetRequest, // reset retrieves the current state of the blockchain and ensures the content // of the transaction pool is valid with regard to the chain state. -func (pool *LegacyPool) reset(oldHead, newHead *types.Header) { +func (pool *LegacyPool) reset(oldHead, newHead *types.Header) map[common.Address]bool { // If we're reorging an old state, reinject all dropped transactions var reinject types.Transactions - + demoteAddrs := make(map[common.Address]bool) + collectDemoteAddrs := func(txs types.Transactions) { + if demoteAddrs != nil { + for _, tx := range txs { + addr, _ := types.Sender(pool.signer, tx) + demoteAddrs[addr] = true + } + } + } if oldHead != nil && oldHead.Hash() != newHead.ParentHash { // If the reorg is too deep, avoid doing it (will happen during fast sync) oldNum := oldHead.Number.Uint64() @@ -1351,6 +1360,7 @@ func (pool *LegacyPool) reset(oldHead, newHead *types.Header) { if depth := uint64(math.Abs(float64(oldNum) - float64(newNum))); depth > 64 { log.Debug("Skipping deep transaction reorg", "depth", depth) + demoteAddrs = nil // do a deep txPool reorg } else { // Reorg seems shallow enough to pull in all transactions into memory var ( @@ -1366,7 +1376,7 @@ func (pool *LegacyPool) reset(oldHead, newHead *types.Header) { // If we reorged to a same or higher number, then it's not a case of setHead log.Warn("Transaction pool reset with missing old head", "old", oldHead.Hash(), "oldnum", oldNum, "new", newHead.Hash(), "newnum", newNum) - return + return nil } // If the reorg ended up on a lower number, it's indicative of setHead being the cause log.Debug("Skipping transaction reset caused by setHead", @@ -1379,33 +1389,33 @@ func (pool *LegacyPool) reset(oldHead, newHead *types.Header) { // reorg caused by sync-reversion or explicit sethead back to an // earlier block. log.Warn("Transaction pool reset with missing new head", "number", newHead.Number, "hash", newHead.Hash()) - return + return nil } var discarded, included types.Transactions for rem.NumberU64() > add.NumberU64() { discarded = append(discarded, rem.Transactions()...) if rem = pool.chain.GetBlock(rem.ParentHash(), rem.NumberU64()-1); rem == nil { log.Error("Unrooted old chain seen by tx pool", "block", oldHead.Number, "hash", oldHead.Hash()) - return + return nil } } for add.NumberU64() > rem.NumberU64() { included = append(included, add.Transactions()...) if add = pool.chain.GetBlock(add.ParentHash(), add.NumberU64()-1); add == nil { log.Error("Unrooted new chain seen by tx pool", "block", newHead.Number, "hash", newHead.Hash()) - return + return nil } } for rem.Hash() != add.Hash() { discarded = append(discarded, rem.Transactions()...) if rem = pool.chain.GetBlock(rem.ParentHash(), rem.NumberU64()-1); rem == nil { log.Error("Unrooted old chain seen by tx pool", "block", oldHead.Number, "hash", oldHead.Hash()) - return + return nil } included = append(included, add.Transactions()...) if add = pool.chain.GetBlock(add.ParentHash(), add.NumberU64()-1); add == nil { log.Error("Unrooted new chain seen by tx pool", "block", newHead.Number, "hash", newHead.Hash()) - return + return nil } } lost := make([]*types.Transaction, 0, len(discarded)) @@ -1415,26 +1425,31 @@ func (pool *LegacyPool) reset(oldHead, newHead *types.Header) { } } reinject = lost + collectDemoteAddrs(discarded) + collectDemoteAddrs(included) } } } // Initialize the internal state to the current head if newHead == nil { newHead = pool.chain.CurrentBlock() // Special case during testing + demoteAddrs = nil } statedb, err := pool.chain.StateAt(newHead.Root) if err != nil { log.Error("Failed to reset txpool state", "err", err) - return + return nil } pool.currentHead.Store(newHead) pool.currentState = statedb pool.pendingNonces = newNoncer(statedb) + collectDemoteAddrs(pool.chain.GetBlock(newHead.Hash(), newHead.Number.Uint64()).Transactions()) // Inject any transactions discarded due to reorgs log.Debug("Reinjecting stale transactions", "count", len(reinject)) core.SenderCacher.Recover(pool.signer, reinject) pool.addTxsLocked(reinject, false) + return demoteAddrs } // promoteExecutables moves transactions that have become processable from the @@ -1646,10 +1661,14 @@ func (pool *LegacyPool) truncateQueue() { // Note: transactions are not marked as removed in the priced list because re-heaping // is always explicitly triggered by SetBaseFee and it would be unnecessary and wasteful // to trigger a re-heap is this function -func (pool *LegacyPool) demoteUnexecutables() { +func (pool *LegacyPool) demoteUnexecutables(demoteAddrs map[common.Address]bool) { // Iterate over all accounts and demote any non-executable transactions gasLimit := pool.currentHead.Load().GasLimit for addr, list := range pool.pending { + if demoteAddrs != nil && !demoteAddrs[addr] { + continue + } + nonce := pool.currentState.GetNonce(addr) // Drop all transactions that are deemed too old (low nonce) diff --git a/core/txpool/legacypool/legacypool_test.go b/core/txpool/legacypool/legacypool_test.go index 39673d176db9..e4d929046ce5 100644 --- a/core/txpool/legacypool/legacypool_test.go +++ b/core/txpool/legacypool/legacypool_test.go @@ -2551,7 +2551,7 @@ func benchmarkPendingDemotion(b *testing.B, size int) { // Benchmark the speed of pool validation b.ResetTimer() for i := 0; i < b.N; i++ { - pool.demoteUnexecutables() + pool.demoteUnexecutables(nil) } }