fix(core/txpool): fix Sync deadlock and lost head events in pool loop

Two CI flakes in the txpool reset synchronization surfaced as a 45-minute
test timeout on pool.Sync() and sporadic "nonce too high" rejections.

Sync() could block forever: the resetDone branch notified the waiter with
a non-blocking send and then unconditionally cleared it. When Sync() had
not yet reached its receive (a scheduling gap after it sent its waiter),
the notification was dropped, leaving the waiter never signaled. Restore
the blocking send used upstream, which cannot lose the notification.

Chain head events could be lost: the loop subscribed to head events only
after its goroutine started, so an InsertChain event emitted right after
New could be missed. The pool then reset against a stale head, keeping
pending nonces at genesis and rejecting fresh transactions as
ErrNonceTooHigh. Move the subscription into New so events are captured
from the start.

Stabilize TestResubmit and TestJournal by syncing the pool after Add
before reading its content.

Author: gzliudan

core/txpool/locals/tx_tracker_test.go

@@ -130,6 +130,9 @@ func TestResubmit(t *testing.T) {
 	txsA := txs[:len(txs)/2]
 	txsB := txs[len(txs)/2:]
 	env.pool.Add(txsA, true)
+	if err := env.pool.Sync(); err != nil {
+		t.Fatalf("Failed to sync the txpool: %v", err)
+	}
 
 	pending, queued := env.pool.ContentFrom(address)
 	if len(pending) != len(txsA) || len(queued) != 0 {
@@ -163,6 +166,9 @@ func TestJournal(t *testing.T) {
 	txsA := txs[:len(txs)/2]
 	txsB := txs[len(txs)/2:]
 	env.pool.Add(txsA, true)
+	if err := env.pool.Sync(); err != nil {
+		t.Fatalf("Failed to sync the txpool: %v", err)
+	}
 
 	pending, queued := env.pool.ContentFrom(address)
 	if len(pending) != len(txsA) || len(queued) != 0 {

core/txpool/txpool.go

@@ -76,6 +76,9 @@ type TxPool struct {
 	quit chan chan error         // Quit channel to tear down the head updater
 	term chan struct{}           // Termination channel to detect a closed pool
 
+	newHeadCh  chan core.ChainHeadEvent // Channel of chain head events to trigger subpool resets
+	newHeadSub event.Subscription       // Subscription to the chain head events
+
 	sync chan chan error // Testing / simulator channel to block until internal reset is done
 }
 
@@ -105,13 +108,17 @@ func New(gasTip uint64, chain BlockChain, subpools []SubPool) (*TxPool, error) {
 		return nil, err
 	}
 	pool := &TxPool{
-		subpools: subpools,
-		chain:    chain,
-		state:    statedb,
-		quit:     make(chan chan error),
-		term:     make(chan struct{}),
-		sync:     make(chan chan error),
+		subpools:  subpools,
+		chain:     chain,
+		state:     statedb,
+		quit:      make(chan chan error),
+		term:      make(chan struct{}),
+		sync:      make(chan chan error),
+		newHeadCh: make(chan core.ChainHeadEvent),
 	}
+	// Subscribe to chain head events synchronously, before any block is inserted,
+	// so head events emitted right after initialization are not lost.
+	pool.newHeadSub = chain.SubscribeChainHeadEvent(pool.newHeadCh)
 	reserver := NewReservationTracker()
 	for i, subpool := range subpools {
 		if err := subpool.Init(gasTip, head, reserver.NewHandle(i)); err != nil {
@@ -157,12 +164,11 @@ func (p *TxPool) loop(head *types.Header) {
 	// Close the termination marker when the pool stops
 	defer close(p.term)
 
-	// Subscribe to chain head events to trigger subpool resets
-	var (
-		newHeadCh  = make(chan core.ChainHeadEvent)
-		newHeadSub = p.chain.SubscribeChainHeadEvent(newHeadCh)
-	)
-	defer newHeadSub.Unsubscribe()
+	// Consume chain head events to trigger subpool resets. The subscription is
+	// established synchronously in New, so head events emitted immediately after
+	// initialization are delivered instead of being lost.
+	newHeadCh := p.newHeadCh
+	defer p.newHeadSub.Unsubscribe()
 
 	// Track the previous and current head to feed to an idle reset
 	var (
@@ -177,13 +183,12 @@ func (p *TxPool) loop(head *types.Header) {
 		resetForced bool       // Whether a forced reset was requested, only used in simulator mode
 		resetWaiter chan error // Channel waiting on a forced reset, only used in simulator mode
 	)
-	// Notify the live reset waiter without blocking if the txpool is closed.
+	// Notify the live reset waiter when the pool terminates. The send blocks
+	// until Sync() picks it up; it cannot abort via p.term (closed only after
+	// this defer runs), so the notification is guaranteed to be delivered.
 	defer func() {
 		if resetWaiter != nil {
-			select {
-			case resetWaiter <- errors.New("pool already terminated"):
-			default:
-			}
+			resetWaiter <- errors.New("pool already terminated")
 			resetWaiter = nil
 		}
 	}()
@@ -245,12 +250,12 @@ func (p *TxPool) loop(head *types.Header) {
 			// the forced op is still pending. In that case, wait another round
 			// of resets.
 			if resetWaiter != nil && !resetForced {
-				select {
-				case resetWaiter <- nil:
-					// notification delivered
-				default:
-					// no active listener; avoid blocking the event loop
-				}
+				// Block until the waiter receives the notification. Sync() is
+				// guaranteed to be waiting on its waiter channel (it cannot
+				// abort via p.term while this loop is still running), so a
+				// non-blocking send here could drop the notification and leave
+				// Sync() blocked forever.
+				resetWaiter <- nil
 				resetWaiter = nil
 			}