core/txpool, eth: preserve rebroadcast retries and avoid timeout jailing

core/txpool/legacypool/legacypool.go

@@ -338,10 +338,11 @@ type LegacyPool struct {
 	filteredAddrs map[common.Address]struct{} // Map of addresses to filter
 
 	// Rebroadcast tracking
-	rebroadcastTxFeed  event.Feed // Feed for stuck transaction events
-	rebroadcastAckFeed event.Feed
-	lastRebroadcast    map[common.Hash]time.Time // Track last rebroadcast time per tx hash
-	isTxPrivate        func(common.Hash) bool
+	rebroadcastTxFeed      event.Feed // Feed for stuck transaction events
+	rebroadcastAckFeed     event.Feed
+	lastRebroadcast        map[common.Hash]time.Time // Track last queue acceptance per tx hash
+	lastRebroadcastAttempt map[common.Hash]time.Time
+	isTxPrivate            func(common.Hash) bool
 
 	stranded strandedState // Pending chains blocked by an unminable head
 }
@@ -359,22 +360,23 @@ func New(config Config, chain BlockChain, options ...func(pool *LegacyPool)) *Le
 	// Create the transaction pool with its initial settings
 	signer := types.LatestSigner(chain.Config())
 	pool := &LegacyPool{
-		config:          config,
-		chain:           chain,
-		chainconfig:     chain.Config(),
-		signer:          signer,
-		pending:         make(map[common.Address]*list),
-		queue:           newQueue(config, signer),
-		all:             newLookup(),
-		reqResetCh:      make(chan *txpoolResetRequest),
-		reqPromoteCh:    make(chan *accountSet),
-		queueTxEventCh:  make(chan *types.Transaction),
-		reorgDoneCh:     make(chan chan struct{}),
-		reorgShutdownCh: make(chan struct{}),
-		initDoneCh:      make(chan struct{}),
-		filteredAddrs:   make(map[common.Address]struct{}),
-		lastRebroadcast: make(map[common.Hash]time.Time),
-		stranded:        newStrandedState(config),
+		config:                 config,
+		chain:                  chain,
+		chainconfig:            chain.Config(),
+		signer:                 signer,
+		pending:                make(map[common.Address]*list),
+		queue:                  newQueue(config, signer),
+		all:                    newLookup(),
+		reqResetCh:             make(chan *txpoolResetRequest),
+		reqPromoteCh:           make(chan *accountSet),
+		queueTxEventCh:         make(chan *types.Transaction),
+		reorgDoneCh:            make(chan chan struct{}),
+		reorgShutdownCh:        make(chan struct{}),
+		initDoneCh:             make(chan struct{}),
+		filteredAddrs:          make(map[common.Address]struct{}),
+		lastRebroadcast:        make(map[common.Hash]time.Time),
+		lastRebroadcastAttempt: make(map[common.Hash]time.Time),
+		stranded:               newStrandedState(config),
 	}
 	pool.priced = newPricedList(pool.all)
 
@@ -550,78 +552,6 @@ func (pool *LegacyPool) SubscribeRebroadcastTransactions(ch chan<- core.StuckTxs
 	return pool.rebroadcastTxFeed.Subscribe(ch)
 }
 
-// identifyStuckTransactions identifies pending transactions that may be stuck
-// and need rebroadcasting. It returns transactions that:
-// - Have been pending longer than RebroadcastInterval but less than RebroadcastMaxAge
-// - Have not been rebroadcast recently (within RebroadcastInterval)
-// - Are immediately executable (gas price meets current requirements)
-//
-// Must be called with pool.mu.RLock held (read lock only - does not modify pool state).
-func (pool *LegacyPool) identifyStuckTransactions() []*types.Transaction {
-	now := time.Now()
-	head := pool.currentHead.Load()
-	if head == nil {
-		return nil
-	}
-
-	// Calculate base fee only if London is enabled and header has base fee
-	var baseFee *big.Int
-	if pool.chainconfig.IsLondon(head.Number) && head.BaseFee != nil {
-		baseFee = eip1559.CalcBaseFee(pool.chainconfig, head)
-	}
-	minTip := pool.gasTip.Load().ToBig()
-
-	var stuckTxs []*types.Transaction
-
-	for _, list := range pool.pending {
-		for _, tx := range list.Flatten() {
-			hash := tx.Hash()
-			age := now.Sub(tx.Time())
-
-			// Skip if too young (hasn't had time to propagate yet)
-			if age < pool.config.RebroadcastInterval {
-				continue
-			}
-
-			// Check rebroadcast history
-			lastTime, wasRebroadcast := pool.lastRebroadcast[hash]
-			if wasRebroadcast {
-				// Skip if recently rebroadcast
-				if now.Sub(lastTime) < pool.config.RebroadcastInterval {
-					continue
-				}
-				// Skip if we've been rebroadcasting this tx for too long (max age applies
-				// only to previously rebroadcast txs - new txs that just became executable
-				// after a base fee drop should still be considered)
-				if pool.config.RebroadcastMaxAge > 0 && age > pool.config.RebroadcastMaxAge {
-					continue
-				}
-			}
-
-			// Skip if not immediately executable (gas price too low for current conditions)
-			// For EIP-1559 transactions, check gas fee cap against base fee
-			if baseFee != nil && tx.GasFeeCap().Cmp(baseFee) < 0 {
-				continue
-			}
-			if tx.GasTipCap().Cmp(minTip) < 0 {
-				continue
-			}
-			if !pool.canRebroadcast(tx) {
-				continue
-			}
-
-			stuckTxs = append(stuckTxs, tx)
-
-			// Enforce batch limit
-			if len(stuckTxs) >= pool.config.RebroadcastBatchSize {
-				return stuckTxs
-			}
-		}
-	}
-
-	return stuckTxs
-}
-
 // SetGasTip updates the minimum gas tip required by the transaction pool for a
 // new transaction, and drops all transactions below this threshold.
 func (pool *LegacyPool) SetGasTip(tip *big.Int) {
@@ -1145,7 +1075,7 @@ func (pool *LegacyPool) add(tx *types.Transaction, async bool) (replaced bool, e
 				pool.priced.Removed(1)
 			}
 			pendingReplaceMeter.Mark(1)
-			delete(pool.lastRebroadcast, old.Hash())
+			pool.forgetRebroadcast(old.Hash())
 		}
 		pool.all.Add(tx)
 		reheapCount := pool.priced.reheaps.Load()
@@ -1240,15 +1170,15 @@ func (pool *LegacyPool) promoteTx(addr common.Address, hash common.Hash, tx *typ
 		pool.all.Remove(hash)
 		pool.priced.Removed(1)
 		pendingDiscardMeter.Mark(1)
-		delete(pool.lastRebroadcast, hash)
+		pool.forgetRebroadcast(hash)
 		return false
 	}
 	// Otherwise discard any previous transaction and mark this
 	if old != nil {
 		pool.all.Remove(old.Hash())
 		pool.priced.Removed(1)
 		pendingReplaceMeter.Mark(1)
-		delete(pool.lastRebroadcast, old.Hash())
+		pool.forgetRebroadcast(old.Hash())
 	} else {
 		// Nothing was replaced, bump the pending counter
 		pendingGauge.Inc(1)
@@ -1466,7 +1396,7 @@ func (pool *LegacyPool) removeTx(hash common.Hash, outofbound bool, unreserve bo
 		pool.priced.Removed(1)
 	}
 	// Clean up rebroadcast tracking
-	delete(pool.lastRebroadcast, hash)
+	pool.forgetRebroadcast(hash)
 	// Remove the transaction from the pending lists and reset the account nonce
 	if pending := pool.pending[addr]; pending != nil {
 		if removed, invalids := pending.Remove(tx); removed {
@@ -1852,7 +1782,7 @@ func (pool *LegacyPool) promoteExecutables(accounts []common.Address) []*types.T
 	// remove all removable transactions
 	for _, hash := range dropped {
 		pool.all.Remove(hash)
-		delete(pool.lastRebroadcast, hash)
+		pool.forgetRebroadcast(hash)
 	}
 	pool.priced.Removed(len(dropped))
 
@@ -1909,7 +1839,7 @@ func (pool *LegacyPool) truncatePending() {
 						// Drop the transaction from the global pools too
 						hash := tx.Hash()
 						pool.all.Remove(hash)
-						delete(pool.lastRebroadcast, hash)
+						pool.forgetRebroadcast(hash)
 
 						// Update the account nonce to the dropped transaction
 						pool.pendingNonces.setIfLower(offenders[i], tx.Nonce())
@@ -1935,7 +1865,7 @@ func (pool *LegacyPool) truncatePending() {
 					// Drop the transaction from the global pools too
 					hash := tx.Hash()
 					pool.all.Remove(hash)
-					delete(pool.lastRebroadcast, hash)
+					pool.forgetRebroadcast(hash)
 
 					// Update the account nonce to the dropped transaction
 					pool.pendingNonces.setIfLower(addr, tx.Nonce())
@@ -1957,7 +1887,7 @@ func (pool *LegacyPool) truncateQueue() {
 	// Remove all removable transactions from the lookup and global price list
 	for _, hash := range removed {
 		pool.all.Remove(hash)
-		delete(pool.lastRebroadcast, hash)
+		pool.forgetRebroadcast(hash)
 	}
 	pool.priced.Removed(len(removed))
 
@@ -1990,15 +1920,15 @@ func (pool *LegacyPool) demoteUnexecutables() {
 		for _, tx := range olds {
 			hash := tx.Hash()
 			pool.all.Remove(hash)
-			delete(pool.lastRebroadcast, hash)
+			pool.forgetRebroadcast(hash)
 			log.Trace("Removed old pending transaction", "hash", hash)
 		}
 		// Drop all transactions that are too costly (low balance or out of gas), and queue any invalids back for later
 		drops, invalids := list.Filter(pool.currentState.GetBalance(addr), gasLimit)
 		for _, tx := range drops {
 			hash := tx.Hash()
 			pool.all.Remove(hash)
-			delete(pool.lastRebroadcast, hash)
+			pool.forgetRebroadcast(hash)
 			log.Trace("Removed unpayable pending transaction", "hash", hash)
 		}
 		pendingNofundsMeter.Mark(int64(len(drops)))
@@ -2016,7 +1946,7 @@ func (pool *LegacyPool) demoteUnexecutables() {
 		for _, tx := range txConditionalsRemoved {
 			hash := tx.Hash()
 			pool.all.Remove(hash)
-			delete(pool.lastRebroadcast, hash)
+			pool.forgetRebroadcast(hash)
 			log.Trace("Removed invalid conditional transaction", "hash", hash)
 		}
 

core/txpool/legacypool/legacypool_stranded.go

@@ -129,7 +129,7 @@ func (pool *LegacyPool) dropPendingAccount(addr common.Address, list *list) {
 	for _, tx := range drops {
 		hash := tx.Hash()
 		pool.all.Remove(hash)
-		delete(pool.lastRebroadcast, hash)
+		pool.forgetRebroadcast(hash)
 	}
 	delete(pool.pending, addr)
 	pool.pendingNonces.setIfLower(addr, first)

core/txpool/legacypool/rebroadcast.go

@@ -45,6 +45,7 @@ func (pool *LegacyPool) publishRebroadcastTransactions(txs []*types.Transaction)
 	explicit := pool.rebroadcastAckFeed.Send(core.StuckTxsEvent{Txs: txs})
 	legacy := pool.rebroadcastTxFeed.Send(core.StuckTxsEvent{Txs: txs})
 	if explicit == 0 || legacy > 0 {
+		pool.recordRebroadcastAttempts(txs)
 		hashes := make([]common.Hash, len(txs))
 		for i, tx := range txs {
 			hashes[i] = tx.Hash()
@@ -90,3 +91,20 @@ func (pool *LegacyPool) rebroadcastAcknowledgement(txs []*types.Transaction, cou
 		}
 	}
 }
+
+// Must be called with pool.mu held, alongside removal from the pool.
+func (pool *LegacyPool) forgetRebroadcast(hash common.Hash) {
+	delete(pool.lastRebroadcast, hash)
+	delete(pool.lastRebroadcastAttempt, hash)
+}
+
+func (pool *LegacyPool) recordRebroadcastAttempts(txs []*types.Transaction) {
+	pool.mu.Lock()
+	defer pool.mu.Unlock()
+	now := time.Now()
+	for _, tx := range txs {
+		if pool.all.Get(tx.Hash()) == tx {
+			pool.lastRebroadcastAttempt[tx.Hash()] = now
+		}
+	}
+}

core/txpool/legacypool/rebroadcast_attempt.go

@@ -0,0 +1,48 @@
+// Copyright 2026 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package legacypool
+
+import (
+	"time"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+)
+
+// RebroadcastAttempt records actual queue offers independently of acceptance.
+// The batch owns one attempt per hash, even when several peers refuse it.
+func (pool *LegacyPool) RebroadcastAttempt(txs []*types.Transaction) func([]common.Hash) {
+	remaining := make(map[common.Hash]*types.Transaction, len(txs))
+	for _, tx := range txs {
+		remaining[tx.Hash()] = tx
+	}
+	return func(hashes []common.Hash) {
+		pool.mu.Lock()
+		defer pool.mu.Unlock()
+		now := time.Now()
+		for _, hash := range hashes {
+			tx := remaining[hash]
+			if tx == nil {
+				continue
+			}
+			delete(remaining, hash)
+			if pool.all.Get(hash) == tx {
+				pool.lastRebroadcastAttempt[hash] = now
+			}
+		}
+	}
+}