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.
This commit is contained in:
Ömer Faruk Irmak 2024-09-20 15:46:09 +03:00
parent b018da9d02
commit 6baf9fdf0b
2 changed files with 32 additions and 13 deletions

View File

@ -1272,9 +1272,10 @@ func (pool *LegacyPool) runReorg(done chan struct{}, reset *txpoolResetRequest,
promoteAddrs = dirtyAccounts.flatten() promoteAddrs = dirtyAccounts.flatten()
} }
pool.mu.Lock() pool.mu.Lock()
var demoteAddrs map[common.Address]bool
if reset != nil { if reset != nil {
// Reset from the old head to the new, rescheduling any reorged transactions // 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 // Nonces were reset, discard any events that became stale
for addr := range events { 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 // remove any transaction that has been included in the block or was invalidated
// because of another transaction (e.g. higher gas price). // because of another transaction (e.g. higher gas price).
if reset != nil { if reset != nil {
pool.demoteUnexecutables() pool.demoteUnexecutables(demoteAddrs)
if reset.newHead != nil { if reset.newHead != nil {
if pool.chainconfig.IsLondon(new(big.Int).Add(reset.newHead.Number, big.NewInt(1))) { if pool.chainconfig.IsLondon(new(big.Int).Add(reset.newHead.Number, big.NewInt(1))) {
pendingBaseFee := eip1559.CalcBaseFee(pool.chainconfig, reset.newHead) 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 // reset retrieves the current state of the blockchain and ensures the content
// of the transaction pool is valid with regard to the chain state. // 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 // If we're reorging an old state, reinject all dropped transactions
var reinject types.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 oldHead != nil && oldHead.Hash() != newHead.ParentHash {
// If the reorg is too deep, avoid doing it (will happen during fast sync) // If the reorg is too deep, avoid doing it (will happen during fast sync)
oldNum := oldHead.Number.Uint64() 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 { if depth := uint64(math.Abs(float64(oldNum) - float64(newNum))); depth > 64 {
log.Debug("Skipping deep transaction reorg", "depth", depth) log.Debug("Skipping deep transaction reorg", "depth", depth)
demoteAddrs = nil // do a deep txPool reorg
} else { } else {
// Reorg seems shallow enough to pull in all transactions into memory // Reorg seems shallow enough to pull in all transactions into memory
var ( 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 // 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", log.Warn("Transaction pool reset with missing old head",
"old", oldHead.Hash(), "oldnum", oldNum, "new", newHead.Hash(), "newnum", newNum) "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 // 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", 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 // reorg caused by sync-reversion or explicit sethead back to an
// earlier block. // earlier block.
log.Warn("Transaction pool reset with missing new head", "number", newHead.Number, "hash", newHead.Hash()) log.Warn("Transaction pool reset with missing new head", "number", newHead.Number, "hash", newHead.Hash())
return return nil
} }
var discarded, included types.Transactions var discarded, included types.Transactions
for rem.NumberU64() > add.NumberU64() { for rem.NumberU64() > add.NumberU64() {
discarded = append(discarded, rem.Transactions()...) discarded = append(discarded, rem.Transactions()...)
if rem = pool.chain.GetBlock(rem.ParentHash(), rem.NumberU64()-1); rem == nil { 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()) log.Error("Unrooted old chain seen by tx pool", "block", oldHead.Number, "hash", oldHead.Hash())
return return nil
} }
} }
for add.NumberU64() > rem.NumberU64() { for add.NumberU64() > rem.NumberU64() {
included = append(included, add.Transactions()...) included = append(included, add.Transactions()...)
if add = pool.chain.GetBlock(add.ParentHash(), add.NumberU64()-1); add == nil { 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()) log.Error("Unrooted new chain seen by tx pool", "block", newHead.Number, "hash", newHead.Hash())
return return nil
} }
} }
for rem.Hash() != add.Hash() { for rem.Hash() != add.Hash() {
discarded = append(discarded, rem.Transactions()...) discarded = append(discarded, rem.Transactions()...)
if rem = pool.chain.GetBlock(rem.ParentHash(), rem.NumberU64()-1); rem == nil { 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()) log.Error("Unrooted old chain seen by tx pool", "block", oldHead.Number, "hash", oldHead.Hash())
return return nil
} }
included = append(included, add.Transactions()...) included = append(included, add.Transactions()...)
if add = pool.chain.GetBlock(add.ParentHash(), add.NumberU64()-1); add == nil { 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()) 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)) lost := make([]*types.Transaction, 0, len(discarded))
@ -1415,26 +1425,31 @@ func (pool *LegacyPool) reset(oldHead, newHead *types.Header) {
} }
} }
reinject = lost reinject = lost
collectDemoteAddrs(discarded)
collectDemoteAddrs(included)
} }
} }
} }
// Initialize the internal state to the current head // Initialize the internal state to the current head
if newHead == nil { if newHead == nil {
newHead = pool.chain.CurrentBlock() // Special case during testing newHead = pool.chain.CurrentBlock() // Special case during testing
demoteAddrs = nil
} }
statedb, err := pool.chain.StateAt(newHead.Root) statedb, err := pool.chain.StateAt(newHead.Root)
if err != nil { if err != nil {
log.Error("Failed to reset txpool state", "err", err) log.Error("Failed to reset txpool state", "err", err)
return return nil
} }
pool.currentHead.Store(newHead) pool.currentHead.Store(newHead)
pool.currentState = statedb pool.currentState = statedb
pool.pendingNonces = newNoncer(statedb) pool.pendingNonces = newNoncer(statedb)
collectDemoteAddrs(pool.chain.GetBlock(newHead.Hash(), newHead.Number.Uint64()).Transactions())
// Inject any transactions discarded due to reorgs // Inject any transactions discarded due to reorgs
log.Debug("Reinjecting stale transactions", "count", len(reinject)) log.Debug("Reinjecting stale transactions", "count", len(reinject))
core.SenderCacher.Recover(pool.signer, reinject) core.SenderCacher.Recover(pool.signer, reinject)
pool.addTxsLocked(reinject, false) pool.addTxsLocked(reinject, false)
return demoteAddrs
} }
// promoteExecutables moves transactions that have become processable from the // 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 // 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 // is always explicitly triggered by SetBaseFee and it would be unnecessary and wasteful
// to trigger a re-heap is this function // 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 // Iterate over all accounts and demote any non-executable transactions
gasLimit := pool.currentHead.Load().GasLimit gasLimit := pool.currentHead.Load().GasLimit
for addr, list := range pool.pending { for addr, list := range pool.pending {
if demoteAddrs != nil && !demoteAddrs[addr] {
continue
}
nonce := pool.currentState.GetNonce(addr) nonce := pool.currentState.GetNonce(addr)
// Drop all transactions that are deemed too old (low nonce) // Drop all transactions that are deemed too old (low nonce)

View File

@ -2551,7 +2551,7 @@ func benchmarkPendingDemotion(b *testing.B, size int) {
// Benchmark the speed of pool validation // Benchmark the speed of pool validation
b.ResetTimer() b.ResetTimer()
for i := 0; i < b.N; i++ { for i := 0; i < b.N; i++ {
pool.demoteUnexecutables() pool.demoteUnexecutables(nil)
} }
} }