fix(core/txpool,eth): prevent a stalled tx subscriber from freezing the pool

core/txpool/legacypool/legacypool.go

@@ -244,6 +244,7 @@ type LegacyPool struct {
 	chain       BlockChain
 	gasTip      atomic.Pointer[uint256.Int]
 	txFeed      event.Feed
+	scope       event.SubscriptionScope // Subscription scope to unsubscribe all on shutdown
 	signer      types.Signer
 	mu          sync.RWMutex
 
@@ -293,7 +294,7 @@ func New(config Config, chain BlockChain) *LegacyPool {
 		all:              newLookup(),
 		reqResetCh:       make(chan *txpoolResetRequest),
 		reqPromoteCh:     make(chan *accountSet),
-		queueTxEventCh:   make(chan *types.Transaction),
+		queueTxEventCh:   make(chan *types.Transaction, 1),
 		reorgDoneCh:      make(chan chan struct{}),
 		reorgShutdownCh:  make(chan struct{}),
 		initDoneCh:       make(chan struct{}),
@@ -393,6 +394,10 @@ func (pool *LegacyPool) loop() {
 func (pool *LegacyPool) Close() error {
 	// Terminate the pool reorger and return
 	close(pool.reorgShutdownCh)
+	// Unsubscribe anyone still listening for tx events. This also wakes up a
+	// runReorg that may be blocked in txFeed.Send because a subscriber stopped
+	// draining, allowing scheduleReorgLoop to observe the shutdown and return.
+	pool.scope.Close()
 	pool.wg.Wait()
 
 	log.Info("Transaction pool stopped")
@@ -413,7 +418,7 @@ func (pool *LegacyPool) SubscribeTransactions(ch chan<- core.NewTxsEvent, reorgs
 	// hard to separate newly discovered transactions from resurrected ones. This
 	// is because the new txs are added to the queue, resurrected ones too and
 	// reorgs run lazily, so separating the two would need a marker.
-	return pool.txFeed.Subscribe(ch)
+	return txpool.TrackOrTerminated(&pool.scope, pool.txFeed.Subscribe(ch))
 }
 
 // SetGasTip updates the minimum gas tip required by the transaction pool for a
@@ -986,7 +991,7 @@ func (pool *LegacyPool) promoteSpecialTx(addr common.Address, tx *types.Transact
 	// Set the potentially new pending nonce and notify any subsystems of the new tx
 	pool.queue.bump(addr)
 	pool.pendingNonces.set(addr, tx.Nonce()+1)
-	pool.txFeed.Send(core.NewTxsEvent{Txs: []*types.Transaction{tx}})
+	pool.queueTxEvent(tx)
 	return true, nil
 }
 
@@ -1221,6 +1226,10 @@ func (pool *LegacyPool) requestPromoteExecutables(set *accountSet) chan struct{}
 }
 
 // queueTxEvent enqueues a transaction event to be sent in the next reorg run.
+// The channel is size-1 so callers that enqueue while holding the pool write
+// lock (e.g. promoteSpecialTx) are decoupled from scheduleReorgLoop's select
+// loop and never block past the first pending event; during shutdown the send
+// falls through to reorgShutdownCh and returns immediately.
 func (pool *LegacyPool) queueTxEvent(tx *types.Transaction) {
 	select {
 	case pool.queueTxEventCh <- tx:
@@ -1276,16 +1285,26 @@ func (pool *LegacyPool) scheduleReorgLoop() {
 			pool.reorgDoneCh <- nextDone
 
 		case tx := <-pool.queueTxEventCh:
-			// Queue up the event, but don't schedule a reorg. It's up to the caller to
-			// request one later if they want the events sent.
+			// Queue up the event, but don't schedule a reorg unless the pool is
+			// idle: callers that queued an event without requesting a reorg
+			// (e.g. a special tx promoted straight to pending) still expect it
+			// delivered, so schedule a run if none is running or pending.
 			addr, _ := types.Sender(pool.signer, tx)
 			if _, ok := queuedEvents[addr]; !ok {
 				queuedEvents[addr] = NewSortedMap()
 			}
 			queuedEvents[addr].Put(tx)
+			if curDone == nil && !launchNextRun {
+				launchNextRun = true
+			}
 
 		case <-curDone:
 			curDone = nil
+			// Deliver any events queued while the run was active: schedule one
+			// more run so they don't wait indefinitely on an idle chain.
+			if len(queuedEvents) > 0 && !launchNextRun {
+				launchNextRun = true
+			}
 
 		case <-pool.reorgShutdownCh:
 			// Wait for current run to finish.

core/txpool/legacypool/legacypool_test.go

@@ -3613,3 +3613,187 @@ func TestSetGasPrice(t *testing.T) {
 		})
 	}
 }
+
+// TestSpecialTxPromotionDoesNotBlockOnTxFeed reproduces the mainnet freeze: promoting a
+// special transaction delivered its NewTxsEvent while holding the pool write lock, so a
+// subscriber that stopped draining wedged the pool itself and, through it, every peer
+// goroutine that wanted to add or read transactions.
+func TestSpecialTxPromotionDoesNotBlockOnTxFeed(t *testing.T) {
+	pool, key := setupPool()
+	defer pool.Close()
+
+	pool.SetSigner(func(common.Address) bool { return true })
+	testAddBalance(pool, crypto.PubkeyToAddress(key.PublicKey), big.NewInt(1_000_000_000_000_000_000))
+
+	// Subscriber that never reads, modelling a stalled txBroadcastLoop.
+	sink := make(chan core.NewTxsEvent)
+	sub := pool.SubscribeTransactions(sink, false)
+	defer sub.Unsubscribe()
+
+	gasPrice := new(big.Int).Add(new(big.Int).Set(common.MinGasPrice), big.NewInt(1))
+	specialTx, err := types.SignTx(types.NewTransaction(0, common.BlockSignersBinary, big.NewInt(1), 100000, gasPrice, nil), types.HomesteadSigner{}, key)
+	if err != nil {
+		t.Fatalf("failed to sign special tx: %v", err)
+	}
+	if !specialTx.IsSpecialTransaction() {
+		t.Fatal("test setup: transaction is not special")
+	}
+
+	added := make(chan error, 1)
+	go func() {
+		added <- pool.Add([]*types.Transaction{specialTx}, false)[0]
+	}()
+	select {
+	case err := <-added:
+		if err != nil {
+			t.Fatalf("failed to add special tx: %v", err)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("Add blocked: the special tx event is delivered while holding the pool lock")
+	}
+
+	// Rigidity check: the special tx must actually be promoted to pending, not just
+	// accepted into the queue. A regression that skipped promotion would let this test
+	// pass trivially (Add returns, the lock is free) while the tx never reached pending.
+	promoted := false
+	deadline := time.Now().Add(2 * time.Second)
+	for !promoted && time.Now().Before(deadline) {
+		pending, _ := pool.Content()
+		for _, txs := range pending {
+			for _, ptx := range txs {
+				if ptx.Hash() == specialTx.Hash() {
+					promoted = true
+					break
+				}
+			}
+			if promoted {
+				break
+			}
+		}
+		if !promoted {
+			time.Sleep(10 * time.Millisecond)
+		}
+	}
+	if !promoted {
+		t.Fatal("special tx was accepted but never promoted to pending")
+	}
+
+	usable := make(chan struct{})
+	go func() {
+		defer close(usable)
+		pool.Stats()
+	}()
+	select {
+	case <-usable:
+	case <-time.After(5 * time.Second):
+		t.Fatal("pool lock still held after promoting a special tx")
+	}
+}
+
+// TestLegacyPoolCloseUnblocksStalledSubscriber locks in the L1 guarantee: a subscriber
+// that stops draining its channel leaves runReorg blocked in txFeed.Send, so close(done)
+// never fires and LegacyPool.Close (wg.Wait) would hang forever, forcing a hard SIGKILL.
+// Closing the subscription scope must remove the stuck subscriber and let Send return, so
+// Close completes without hanging.
+func TestLegacyPoolCloseUnblocksStalledSubscriber(t *testing.T) {
+	pool, key := setupPool()
+
+	// LegacyPool.Close is not idempotent (close(reorgShutdownCh) panics on the
+	// second call), so guard it: the deferred teardown closes the pool on any
+	// early failure path, while the body's goroutine uses the same Once.
+	var closeOnce sync.Once
+	defer closeOnce.Do(func() { pool.Close() })
+
+	pool.SetSigner(func(common.Address) bool { return true })
+	testAddBalance(pool, crypto.PubkeyToAddress(key.PublicKey), big.NewInt(1_000_000_000_000_000_000))
+
+	// Stalled subscriber: never read, never explicitly unsubscribe (Close does it via
+	// the subscription scope).
+	sink := make(chan core.NewTxsEvent)
+	pool.SubscribeTransactions(sink, false)
+
+	gasPrice := new(big.Int).Add(new(big.Int).Set(common.MinGasPrice), big.NewInt(1))
+	tx, err := types.SignTx(types.NewTransaction(0, common.Address{0x01}, big.NewInt(1), 21000, gasPrice, nil), types.HomesteadSigner{}, key)
+	if err != nil {
+		t.Fatalf("failed to sign tx: %v", err)
+	}
+
+	// A sync Add waits for the reorg to finish, which requires txFeed.Send to return.
+	// With a stalled subscriber Send never returns, so Add must block here: this proves
+	// the reorg is wedged (the pre-L1 behaviour), which is exactly the state Close must
+	// survive.
+	added := make(chan error, 1)
+	go func() {
+		added <- pool.Add([]*types.Transaction{tx}, true)[0]
+	}()
+	select {
+	case err := <-added:
+		// An early return here means either the reorg was not wedged (the fix
+		// already worked) or the tx was rejected for an unrelated reason, so
+		// surface the actual error instead of assuming the wedge.
+		t.Fatalf("sync Add returned although the subscriber is stalled (expected Send to block, err=%v)", err)
+	case <-time.After(2 * time.Second):
+		// Confirmed: runReorg is stuck in txFeed.Send.
+	}
+
+	// Close must not hang: scope.Close() removes the stalled subscriber, Send returns,
+	// runReorg finishes and the reorg loop shuts down.
+	closed := make(chan struct{})
+	go func() {
+		closeOnce.Do(func() { pool.Close() })
+		close(closed)
+	}()
+	select {
+	case <-closed:
+		// Close returned as expected.
+	case <-time.After(5 * time.Second):
+		t.Fatal("Close hung: stalled subscriber blocked reorg shutdown (L1 regression)")
+	}
+
+	// The previously blocked Add must now complete, because Send was unblocked by Close.
+	select {
+	case err := <-added:
+		if err != nil {
+			t.Fatalf("Add failed after Close unblocked Send: %v", err)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("Add did not complete after Close unblocked Send")
+	}
+}
+
+// TestQueueTxEventDeliveredWhenIdle locks in the delivery guarantee for events
+// queued without a reorg request: on an idle pool (no reset or promote pending)
+// scheduleReorgLoop must schedule a run to deliver them, otherwise a special tx
+// promoted straight to pending would never be announced.
+func TestQueueTxEventDeliveredWhenIdle(t *testing.T) {
+	pool, key := setupPool()
+	defer pool.Close()
+
+	pool.SetSigner(func(common.Address) bool { return true })
+	testAddBalance(pool, crypto.PubkeyToAddress(key.PublicKey), big.NewInt(1_000_000_000_000_000_000))
+
+	// Draining subscriber.
+	events := make(chan core.NewTxsEvent, 1)
+	sub := pool.SubscribeTransactions(events, false)
+	defer sub.Unsubscribe()
+
+	gasPrice := new(big.Int).Add(new(big.Int).Set(common.MinGasPrice), big.NewInt(1))
+	tx, err := types.SignTx(types.NewTransaction(0, common.Address{0x01}, big.NewInt(1), 21000, gasPrice, nil), types.HomesteadSigner{}, key)
+	if err != nil {
+		t.Fatalf("failed to sign tx: %v", err)
+	}
+
+	// Queue the event without any reorg request: the idle pool must still
+	// deliver it. Before the idle-launch fix this never happened, because
+	// scheduleReorgLoop only queued the event and no run was scheduled.
+	pool.queueTxEvent(tx)
+
+	select {
+	case ev := <-events:
+		if len(ev.Txs) != 1 || ev.Txs[0].Hash() != tx.Hash() {
+			t.Fatalf("unexpected event: got %d txs, first hash %v want %v", len(ev.Txs), ev.Txs[0].Hash(), tx.Hash())
+		}
+	case <-time.After(3 * time.Second):
+		t.Fatal("queued tx event was not delivered on an idle pool")
+	}
+}

core/txpool/subpool.go

@@ -135,7 +135,10 @@ type SubPool interface {
 
 	// SubscribeTransactions subscribes to new transaction events. The subscriber
 	// can decide whether to receive notifications only for newly seen transactions
-	// or also for reorged out ones.
+	// or also for reorged out ones. Implementations must never return a nil
+	// subscription: a pool that is already shutting down yields a terminated
+	// (already unsubscribed) subscription instead, so callers can always wait on
+	// Err() and call Unsubscribe().
 	SubscribeTransactions(ch chan<- core.NewTxsEvent, reorgs bool) event.Subscription
 
 	// Nonce returns the next nonce of an account, with all transactions executable

core/txpool/txpool.go

@@ -135,14 +135,19 @@ func (p *TxPool) Close() error {
 	if err := <-errc; err != nil {
 		errs = append(errs, err)
 	}
+	// Unsubscribe anyone still listening for tx events. This must happen before
+	// terminating the subpools: a subscriber that stopped draining its channel
+	// would otherwise leave a subpool's runReorg blocked in txFeed.Send, and the
+	// subpool Close (wg.Wait) would hang indefinitely. Closing the scope removes
+	// the stuck subscription and unblocks Send.
+	p.subs.Close()
+
 	// Terminate each subpool
 	for _, subpool := range p.subpools {
 		if err := subpool.Close(); err != nil {
 			errs = append(errs, err)
 		}
 	}
-	// Unsubscribe anyone still listening for tx events
-	p.subs.Close()
 
 	if len(errs) > 0 {
 		return fmt.Errorf("subpool close errors: %v", errs)
@@ -390,14 +395,31 @@ func (p *TxPool) Pending(filter PendingFilter) map[common.Address][]*LazyTransac
 	return txs
 }
 
+// TrackOrTerminated tracks sub in scope and returns it. If the scope is
+// already closed (pool shutting down), Track refuses and returns nil; in
+// that case sub is unsubscribed and a terminated subscription is returned
+// instead, so callers can always safely wait on Err() and call Unsubscribe()
+// without leaking the refused subscription.
+func TrackOrTerminated(scope *event.SubscriptionScope, sub event.Subscription) event.Subscription {
+	if tracked := scope.Track(sub); tracked != nil {
+		return tracked
+	}
+	sub.Unsubscribe()
+	return event.NewSubscription(func(quit <-chan struct{}) error { return nil })
+}
+
 // SubscribeTransactions registers a subscription for new transaction events,
 // supporting feeding only newly seen or also resurrected transactions.
 func (p *TxPool) SubscribeTransactions(ch chan<- core.NewTxsEvent, reorgs bool) event.Subscription {
 	subs := make([]event.Subscription, len(p.subpools))
 	for i, subpool := range p.subpools {
+		// TrackOrTerminated never returns nil: a subpool that is already
+		// shutting down yields a terminated subscription instead, keeping
+		// JoinSubscriptions from panicking on a nil entry when the joined
+		// subscription terminates.
 		subs[i] = subpool.SubscribeTransactions(ch, reorgs)
 	}
-	return p.subs.Track(event.JoinSubscriptions(subs...))
+	return TrackOrTerminated(&p.subs, event.JoinSubscriptions(subs...))
 }
 
 // PoolNonce returns the next nonce of an account, with all transactions executable

eth/handler.go

@@ -1043,7 +1043,7 @@ func (pm *ProtocolManager) BroadcastTransactions(txs types.Transactions, propaga
 			log.Trace("Broadcast transaction", "hash", tx.Hash(), "recipients", len(peers))
 		}
 		for peer, hashes := range txset {
-			peer.AsyncSendTransactions(hashes)
+			peer.asyncSendChunked(hashes, false)
 		}
 		return
 	}
@@ -1055,11 +1055,7 @@ func (pm *ProtocolManager) BroadcastTransactions(txs types.Transactions, propaga
 		}
 	}
 	for peer, hashes := range annos {
-		if peer.version >= xdc165 {
-			peer.AsyncSendPooledTransactionHashes(hashes)
-		} else {
-			peer.AsyncSendTransactions(hashes)
-		}
+		peer.asyncSendChunked(hashes, peer.version >= xdc165)
 	}
 }