txnprovider/txpool: wake a caller that goes away while waiting for a block

txnprovider/txpool/pool.go

@@ -725,6 +725,30 @@ func (p *TxPool) best(ctx context.Context, n int, txns *TxnsRlp, onTopOf uint64,
 	availableGas mdgas.FullMdGas,
 	yielded mapset.Set[[32]byte], availableRlpSpace int) (bool, int, error) {
 
+	// sync.Cond has no notion of a context, so a caller that goes away while parked below would
+	// sleep until the next block broadcast the condition, or forever while the chain is stalled.
+	// Broadcasting on cancellation wakes it; every waiter rechecks its own condition anyway. The
+	// broadcast takes the lock so it cannot land between the check below and the wait, which is the
+	// window where a wakeup would be lost.
+	stopWatching, watcherDone := make(chan struct{}), make(chan struct{})
+	go func() {
+		defer close(watcherDone)
+		select {
+		case <-ctx.Done():
+			p.lock.Lock()
+			p.lastSeenCond.Broadcast()
+			p.lock.Unlock()
+		case <-stopWatching:
+		}
+	}()
+	// Joined rather than merely told to stop: with the context already ended the watcher can pick
+	// either case, and one that picked cancellation would otherwise take the lock after this call
+	// had returned. Registered before the lock is taken, so it runs after the lock is released.
+	defer func() {
+		close(stopWatching)
+		<-watcherDone
+	}()
+
 	p.lock.Lock()
 	for last := p.lastSeenBlock.Load(); last < onTopOf; last = p.lastSeenBlock.Load() {
 		select {

txnprovider/txpool/pool_best_lock_test.go

@@ -17,13 +17,17 @@
 package txpool
 
 import (
+	"bytes"
 	"context"
+	"strings"
 	"sync"
 	"testing"
+	"time"
 
 	mapset "github.com/deckarep/golang-set/v2"
 	"github.com/stretchr/testify/require"
 
+	"github.com/erigontech/erigon/common/log/v3"
 	mdgas "github.com/erigontech/erigon/execution/protocol/mdgas"
 )
 
@@ -44,3 +48,59 @@ func TestBestReleasesTheLockWhenTheCallerGivesUpWaitingForABlock(t *testing.T) {
 	require.True(t, p.lock.TryLock(), "best returned holding the pool lock")
 	p.lock.Unlock()
 }
+
+// waitingBuffer records what the pool logged, so a test can tell when a caller has reached the
+// "Waiting for block" trace and is therefore about to park in the condition.
+type waitingBuffer struct {
+	mu  sync.Mutex
+	buf bytes.Buffer
+}
+
+func (w *waitingBuffer) Write(p []byte) (int, error) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	return w.buf.Write(p)
+}
+
+func (w *waitingBuffer) contains(s string) bool {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	return strings.Contains(w.buf.String(), s)
+}
+
+func TestBestReturnsWhenItsCallerGoesAwayWhileWaitingForABlock(t *testing.T) {
+	logs := &waitingBuffer{}
+	logger := log.New()
+	logger.SetHandler(log.StreamHandler(logs, log.LogfmtFormat()))
+
+	lock := &sync.Mutex{}
+	p := &TxPool{
+		lock:         lock,
+		lastSeenCond: sync.NewCond(lock),
+		logger:       logger,
+		pending:      NewPendingSubPool(PendingSubPool, 1),
+		baseFee:      NewSubPool(BaseFeeSubPool, 1),
+		queued:       NewSubPool(QueuedSubPool, 1),
+	}
+	ctx, cancel := context.WithCancel(t.Context())
+
+	returned := make(chan error, 1)
+	go func() {
+		_, _, err := p.best(ctx, 1, &TxnsRlp{}, 1, mdgas.FullMdGas{}, mapset.NewSet[[32]byte](), 0)
+		returned <- err
+	}()
+
+	// Cancel only once it is parked, so this exercises the wait rather than the check before it.
+	require.Eventually(t, func() bool { return logs.contains("Waiting for block") }, 5*time.Second, time.Millisecond)
+	cancel()
+
+	// Nothing else wakes it: no block arrives and the pool is not shutting down. A wait that cannot
+	// observe its caller leaves the builder that made this request holding a read view until one of
+	// those happens, which on a stalled chain is never.
+	select {
+	case err := <-returned:
+		require.ErrorIs(t, err, context.Canceled)
+	case <-time.After(5 * time.Second):
+		t.Fatal("best never returned; only a new block would have woken it")
+	}
+}