diff --git a/txnprovider/txpool/pool.go b/txnprovider/txpool/pool.go index b7fba981023..956c0f8b820 100644 --- a/txnprovider/txpool/pool.go +++ b/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 { diff --git a/txnprovider/txpool/pool_best_lock_test.go b/txnprovider/txpool/pool_best_lock_test.go index b71ea6bce1e..d3b0458b76e 100644 --- a/txnprovider/txpool/pool_best_lock_test.go +++ b/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") + } +}