txpool: fix goroutine leak in Fetch on shutdown

txnprovider/txpool/fetch.go

@@ -50,6 +50,7 @@ type Fetch struct {
 	db                       kv.RwDB
 	stateChangesClient       StateChangesClient
 	wg                       *sync.WaitGroup // used for synchronisation in the tests (nil when not in tests)
+	connectWg                sync.WaitGroup  // tracks goroutines spawned by ConnectCore/ConnectSentries
 	stateChangesParseCtx     *TxnParseContext
 	pooledTxnsParseCtx       *TxnParseContext
 	sentryClients            []sentryproto.SentryClient // sentry clients that will be used for accessing the network
@@ -108,17 +109,22 @@ func (f *Fetch) threadSafeParseStateChangeTxn(cb func(*TxnParseContext) error) e
 // ConnectSentries initialises connection to the sentry
 func (f *Fetch) ConnectSentries() {
 	for i := range f.sentryClients {
+		f.connectWg.Add(2)
 		go func(i int) {
+			defer f.connectWg.Done()
 			f.receiveMessageLoop(f.sentryClients[i])
 		}(i)
 		go func(i int) {
+			defer f.connectWg.Done()
 			f.receivePeerLoop(f.sentryClients[i])
 		}(i)
 	}
 }
 
 func (f *Fetch) ConnectCore() {
+	f.connectWg.Add(1)
 	go func() {
+		defer f.connectWg.Done()
 		for {
 			select {
 			case <-f.ctx.Done():
@@ -127,7 +133,11 @@ func (f *Fetch) ConnectCore() {
 			}
 			if err := f.handleStateChanges(f.ctx, f.stateChangesClient); err != nil {
 				if grpcutil.IsRetryLater(err) || grpcutil.IsEndOfStream(err) {
-					time.Sleep(3 * time.Second)
+					select {
+					case <-f.ctx.Done():
+						return
+					case <-time.After(3 * time.Second):
+					}
 					continue
 				}
 				f.logger.Warn("[txpool.handleStateChanges]", "err", err)
@@ -136,6 +146,9 @@ func (f *Fetch) ConnectCore() {
 	}()
 }
 
+// Wait blocks until all goroutines spawned by ConnectCore and ConnectSentries have exited.
+func (f *Fetch) Wait() { f.connectWg.Wait() }
+
 func (f *Fetch) receiveMessageLoop(sentryClient sentryproto.SentryClient) {
 	for {
 		select {
@@ -145,7 +158,11 @@ func (f *Fetch) receiveMessageLoop(sentryClient sentryproto.SentryClient) {
 		}
 		if _, err := sentryClient.HandShake(f.ctx, &emptypb.Empty{}, grpc.WaitForReady(true)); err != nil {
 			if grpcutil.IsRetryLater(err) || grpcutil.IsEndOfStream(err) {
-				time.Sleep(3 * time.Second)
+				select {
+				case <-f.ctx.Done():
+					return
+				case <-time.After(3 * time.Second):
+				}
 				continue
 			}
 			// Report error and wait more
@@ -154,7 +171,11 @@ func (f *Fetch) receiveMessageLoop(sentryClient sentryproto.SentryClient) {
 		}
 		if err := f.receiveMessage(f.ctx, sentryClient); err != nil {
 			if grpcutil.IsRetryLater(err) || grpcutil.IsEndOfStream(err) {
-				time.Sleep(3 * time.Second)
+				select {
+				case <-f.ctx.Done():
+					return
+				case <-time.After(3 * time.Second):
+				}
 				continue
 			}
 			f.logger.Warn("[txpool.recvMessage]", "err", err)
@@ -484,17 +505,29 @@ func (f *Fetch) receivePeerLoop(sentryClient sentryproto.SentryClient) {
 		}
 		if _, err := sentryClient.HandShake(f.ctx, &emptypb.Empty{}, grpc.WaitForReady(true)); err != nil {
 			if grpcutil.IsRetryLater(err) || grpcutil.IsEndOfStream(err) {
-				time.Sleep(3 * time.Second)
+				select {
+				case <-f.ctx.Done():
+					return
+				case <-time.After(3 * time.Second):
+				}
 				continue
 			}
 			// Report error and wait more
 			f.logger.Warn("[txpool.recvPeers] sentry not ready yet", "err", err)
-			time.Sleep(time.Second)
+			select {
+			case <-f.ctx.Done():
+				return
+			case <-time.After(time.Second):
+			}
 			continue
 		}
 		if err := f.receivePeer(sentryClient); err != nil {
 			if grpcutil.IsRetryLater(err) || grpcutil.IsEndOfStream(err) {
-				time.Sleep(3 * time.Second)
+				select {
+				case <-f.ctx.Done():
+					return
+				case <-time.After(3 * time.Second):
+				}
 				continue
 			}
 

txnprovider/txpool/fetch_test.go

@@ -23,6 +23,7 @@ import (
 	"io"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -477,6 +478,74 @@ func TestNoPenaltyOnInternalDBError(t *testing.T) {
 	require.Error(t, err, "internal DB error should be propagated, not swallowed")
 }
 
+// TestFetchConnectGoroutinesExitOnCancel is a regression test for a goroutine
+// leak where ConnectCore/ConnectSentries spawned fire-and-forget goroutines
+// that used bare time.Sleep in retry loops. After context cancellation, Run()
+// would return via the errgroup while the fetch goroutines continued sleeping
+// through a 3-second backoff, racing with downstream cleanup (DB.Close(), etc.).
+//
+// The fix replaced time.Sleep with select on ctx.Done() and added a WaitGroup
+// so callers can block until all goroutines exit. This test verifies that all
+// goroutines exit promptly (well under 3s) after cancellation.
+func TestFetchConnectGoroutinesExitOnCancel(t *testing.T) {
+	t.Parallel()
+	ctx, cancel := context.WithCancel(t.Context())
+
+	// Mock sentry server: HandShake returns io.EOF, forcing receiveMessageLoop
+	// and receivePeerLoop into retry-with-backoff loops.
+	srv := &retrySentryServer{}
+	sentryClient, err := direct.NewSentryClientDirect(direct.ETH68, srv, nil)
+	require.NoError(t, err)
+
+	// Mock state changes client: StateChanges returns io.EOF, forcing the
+	// ConnectCore goroutine into its retry loop.
+	ctrl := gomock.NewController(t)
+	stateChanges := remoteproto.NewMockKVClient(ctrl)
+	stateChanges.EXPECT().
+		StateChanges(gomock.Any(), gomock.Any(), gomock.Any()).
+		Return(nil, io.EOF).AnyTimes()
+
+	pool := NewMockPool(ctrl)
+
+	fetch := NewFetch(ctx, []sentryproto.SentryClient{sentryClient}, pool, stateChanges, nil, u256.N1, log.New())
+
+	// Start all goroutines: 1 from ConnectCore + 2 from ConnectSentries.
+	fetch.ConnectCore()
+	fetch.ConnectSentries()
+
+	// Let goroutines enter their retry-sleep selects.
+	time.Sleep(50 * time.Millisecond)
+
+	// Cancel the context — goroutines should notice via ctx.Done() and exit.
+	cancel()
+
+	// Wait must return well within 3 seconds. Before the fix, goroutines used
+	// bare time.Sleep(3 * time.Second) and would not notice cancellation until
+	// the sleep completed.
+	done := make(chan struct{})
+	go func() {
+		fetch.Wait()
+		close(done)
+	}()
+
+	select {
+	case <-done:
+		// All goroutines exited promptly.
+	case <-time.After(1 * time.Second):
+		t.Fatal("goroutines did not exit within 1s of context cancellation — likely sleeping through backoff")
+	}
+}
+
+// retrySentryServer is a minimal SentryServer where HandShake always returns
+// io.EOF, forcing Fetch goroutines into their retry-with-backoff loops.
+type retrySentryServer struct {
+	sentryproto.UnimplementedSentryServer
+}
+
+func (s *retrySentryServer) HandShake(context.Context, *emptypb.Empty) (*sentryproto.HandShakeReply, error) {
+	return nil, io.EOF
+}
+
 func testRlps(num int) [][]byte {
 	rlps := make([][]byte, num)
 	for i := 0; i < num; i++ {

txnprovider/txpool/pool.go

@@ -1963,6 +1963,7 @@ func (p *TxPool) Run(ctx context.Context) error {
 	}()
 	p.p2pFetcher.ConnectCore()
 	p.p2pFetcher.ConnectSentries()
+	defer p.p2pFetcher.Wait()
 
 	syncToNewPeersEvery := time.NewTicker(p.cfg.SyncToNewPeersEvery)
 	defer syncToNewPeersEvery.Stop()