ethmonitor: add poc for goroutine leak during shutdown (#197)

* ethmonitor: add poc for goroutine leak during shutdown

* ethmonitor: fixes after code review

* ethmonitor: fix goroutine leak

* rename PoC test to TestMonitorShutdownNoGoroutineLeak

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* ethmonitor: fix goroutine leak

---------

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Author: xiam

ethmonitor/ethmonitor.go

@@ -441,7 +441,10 @@ func (m *Monitor) listenNewHead() <-chan uint64 {
 					return
 
 				case <-time.After(time.Duration(m.pollInterval.Load())):
-					nextBlock <- 0
+					select {
+					case nextBlock <- 0:
+					case <-m.ctx.Done():
+					}
 				}
 			}
 		}
@@ -474,12 +477,24 @@ func (m *Monitor) listenNewHead() <-chan uint64 {
 			if nextBlockNumber == 0 || latestBlockNum > nextBlockNumber {
 				// monitor is behind, so we just push to keep going without
 				// waiting on the nextBlock channel
-				ch <- nextBlockNumber
+				select {
+				case ch <- nextBlockNumber:
+				case <-m.ctx.Done():
+					return
+				}
 				continue
 			} else {
 				// wait for the next block
-				<-nextBlock
-				ch <- latestBlockNum
+				select {
+				case <-nextBlock:
+				case <-m.ctx.Done():
+					return
+				}
+				select {
+				case ch <- latestBlockNum:
+				case <-m.ctx.Done():
+					return
+				}
 			}
 		}
 	}()
@@ -1008,7 +1023,10 @@ func (m *Monitor) publish(ctx context.Context, events Blocks) error {
 	// Publish events existing in the queue
 	pubEvents, ok := m.publishQueue.dequeue(maxBlockNum)
 	if ok {
-		m.publishCh <- pubEvents
+		select {
+		case m.publishCh <- pubEvents:
+		case <-m.ctx.Done():
+		}
 	}
 
 	return nil

ethmonitor/goroutine_leak_poc_test.go

@@ -0,0 +1,66 @@
+package ethmonitor_test
+
+//go:generate mockgen -destination=internal/mocks/mock_provider.go -package=mocks github.com/0xsequence/ethkit/ethrpc RawInterface
+
+import (
+	"context"
+	"fmt"
+	"math/big"
+	"runtime"
+	"testing"
+	"time"
+
+	"github.com/0xsequence/ethkit/ethmonitor"
+	"github.com/0xsequence/ethkit/ethmonitor/internal/mocks"
+	"go.uber.org/mock/gomock"
+)
+
+// TestMonitorShutdownNoGoroutineLeak verifies that the monitor shuts down cleanly without leaking goroutines.
+func TestMonitorShutdownNoGoroutineLeak(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping in short mode")
+	}
+
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	provider := mocks.NewMockRawInterface(ctrl)
+	provider.EXPECT().ChainID(gomock.Any()).Return(big.NewInt(1), nil).AnyTimes()
+	provider.EXPECT().IsStreamingEnabled().Return(false).AnyTimes()
+	provider.EXPECT().RawBlockByNumber(gomock.Any(), gomock.Any()).
+		Return(nil, fmt.Errorf("simulated network error")).AnyTimes()
+
+	baseline := runtime.NumGoroutine()
+
+	opts := ethmonitor.DefaultOptions
+	opts.PollingInterval = 10 * time.Millisecond
+	opts.Timeout = 50 * time.Millisecond
+
+	monitor, err := ethmonitor.NewMonitor(provider, opts)
+	if err != nil {
+		t.Fatalf("failed to create monitor: %v", err)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+
+	done := make(chan error, 1)
+	go func() {
+		done <- monitor.Run(ctx)
+	}()
+
+	time.Sleep(200 * time.Millisecond)
+	cancel()
+
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		t.Fatal("Monitor.Run() didn't return within timeout")
+	}
+
+	runtime.GC()
+	time.Sleep(100 * time.Millisecond)
+
+	if leaked := runtime.NumGoroutine() - baseline; leaked > 0 {
+		t.Fatalf("%d goroutine(s) leaked", leaked)
+	}
+}