From f0129dfc267251d10512d72115278d119a52b321 Mon Sep 17 00:00:00 2001 From: Damilola Edwards Date: Thu, 23 Jul 2026 16:25:47 +0100 Subject: [PATCH] Fix shutdown ordering and WaitGroup race between discovery and search close Server cleanup closed the search runtime before stopping the app, and the proxy client's Stop only signaled its background refresh goroutine to exit rather than waiting for it. A discovery tick already in flight could still call the search runtime's OnDiscover, adding to its WaitGroup at the same time Close was waiting on it, which the WaitGroup docs call misuse and the Go runtime turns into a panic. Even without the panic, Close could block for however long that stray activation took, with no bound. Three changes close this from both ends: cleanup now stops the app before closing the search runtime, the proxy client's Stop joins its background refresh goroutine instead of just signaling it, and the search runtime tracks whether it has started closing so OnDiscover becomes a no-op instead of adding to the WaitGroup once it has. Any of the three would narrow the window; together they close it. --- pkg/proxy/client.go | 23 +++- pkg/proxy/client_test.go | 75 +++++++++++++ .../ondiscover_close_race_test.go | 101 ++++++++++++++++++ pkg/searchruntime/runtime.go | 26 ++++- pkg/server/builder.go | 9 +- 5 files changed, 228 insertions(+), 6 deletions(-) create mode 100644 pkg/searchruntime/ondiscover_close_race_test.go diff --git a/pkg/proxy/client.go b/pkg/proxy/client.go index 9f2e317e..29dbe9a4 100644 --- a/pkg/proxy/client.go +++ b/pkg/proxy/client.go @@ -119,6 +119,12 @@ type proxyClient struct { stopCh chan struct{} stopped bool + // refreshWG tracks the background refresh goroutine so Stop can join it + // instead of merely signaling it, guaranteeing no discovery tick is still + // in flight (and so cannot still invoke OnDiscover) by the time Stop + // returns. + refreshWG sync.WaitGroup + // discovered reports whether at least one datasource discovery has // succeeded. It gates server readiness: until the first successful // discovery the server has no datasources to serve. @@ -200,23 +206,32 @@ func (c *proxyClient) Start(ctx context.Context) error { // Start background refresh if configured. if c.cfg.DiscoveryInterval > 0 { + c.refreshWG.Add(1) + go c.backgroundRefresh() } return nil } -// Stop stops the client. +// Stop stops the client and waits for the background refresh goroutine to +// fully exit before returning, so no discovery tick already in flight can +// still invoke OnDiscover after a caller believes the client has stopped. func (c *proxyClient) Stop(_ context.Context) error { c.mu.Lock() - defer c.mu.Unlock() - if c.stopped { + c.mu.Unlock() + return nil } c.stopped = true close(c.stopCh) + c.mu.Unlock() + + // Released above: backgroundRefresh calls Discover, which takes this same + // lock, so waiting on it while still holding the lock would deadlock. + c.refreshWG.Wait() c.log.Info("Proxy client stopped") @@ -628,6 +643,8 @@ func (c *proxyClient) EnsureAuthenticated(ctx context.Context) error { // backgroundRefresh periodically refreshes datasource information. func (c *proxyClient) backgroundRefresh() { + defer c.refreshWG.Done() + ticker := time.NewTicker(c.cfg.DiscoveryInterval) defer ticker.Stop() diff --git a/pkg/proxy/client_test.go b/pkg/proxy/client_test.go index b7e63798..bbbfa3fe 100644 --- a/pkg/proxy/client_test.go +++ b/pkg/proxy/client_test.go @@ -76,6 +76,81 @@ func TestDiscoverFiresOnDiscoverHook(t *testing.T) { } } +// TestStopWaitsForInFlightBackgroundRefresh verifies Stop blocks until a +// background discovery tick already running its OnDiscover hook has finished, +// and that no further tick fires once Stop has returned. Before the fix, Stop +// only signaled the background goroutine to exit and returned immediately, +// so a caller could believe discovery had fully stopped while a tick was +// still in flight. +func TestStopWaitsForInFlightBackgroundRefresh(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(DatasourcesResponse{}) + })) + t.Cleanup(srv.Close) + + log := logrus.New() + log.SetOutput(io.Discard) + + var hookCalls atomic.Int32 + entered := make(chan struct{}, 1) + release := make(chan struct{}) + + client := NewClient(log, ClientConfig{ + URL: srv.URL, + DiscoveryInterval: 10 * time.Millisecond, + OnDiscover: func() { + n := hookCalls.Add(1) + // Block the second call only (the first background tick), so the + // initial Start-time Discover above completes normally. + if n == 2 { + entered <- struct{}{} + <-release + } + }, + }).(*proxyClient) + + if err := client.Start(context.Background()); err != nil { + t.Fatalf("Start error = %v", err) + } + + <-entered // a background tick's OnDiscover is now blocked inside the hook + + stopDone := make(chan error, 1) + go func() { + stopDone <- client.Stop(context.Background()) + }() + + select { + case <-stopDone: + t.Fatal("Stop returned while a background tick's OnDiscover was still in flight") + case <-time.After(100 * time.Millisecond): + } + + close(release) + + select { + case err := <-stopDone: + if err != nil { + t.Fatalf("Stop error = %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("Stop did not return after the in-flight OnDiscover finished") + } + + callsAtStop := hookCalls.Load() + + // If the background loop were still ticking, this would be long enough to + // see another call. + time.Sleep(50 * time.Millisecond) + + if got := hookCalls.Load(); got != callsAtStop { + t.Fatalf("hookCalls changed after Stop returned: %d -> %d; background refresh should be fully stopped", callsAtStop, got) + } +} + // TestDiscoverNilOnDiscoverIsSafe verifies a nil OnDiscover hook does not // panic the discovery goroutine. func TestDiscoverNilOnDiscoverIsSafe(t *testing.T) { diff --git a/pkg/searchruntime/ondiscover_close_race_test.go b/pkg/searchruntime/ondiscover_close_race_test.go new file mode 100644 index 00000000..2f5f8298 --- /dev/null +++ b/pkg/searchruntime/ondiscover_close_race_test.go @@ -0,0 +1,101 @@ +package searchruntime + +import ( + "context" + "net/http" + "net/http/httptest" + "net/url" + "sync" + "testing" + "time" + + "github.com/sirupsen/logrus" + "github.com/stretchr/testify/require" + + "github.com/ethpandaops/panda/pkg/types" +) + +// noEmbeddingProxyService is a minimal proxy.Service whose embedding probes +// always fail fast (a 404 from a real local server, not a hang), so +// Runtime.activate reliably takes its "embedding not available" early-return +// path on every call. That makes OnDiscover's Add/Done cycle fast enough to +// hammer repeatedly in a race test. +type noEmbeddingProxyService struct { + url string +} + +func newNoEmbeddingProxyService(t *testing.T) *noEmbeddingProxyService { + t.Helper() + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusNotFound) + })) + t.Cleanup(srv.Close) + + return &noEmbeddingProxyService{url: srv.URL} +} + +func (f *noEmbeddingProxyService) Start(context.Context) error { return nil } +func (f *noEmbeddingProxyService) Stop(context.Context) error { return nil } +func (f *noEmbeddingProxyService) URL() string { return f.url } +func (f *noEmbeddingProxyService) Ready() bool { return true } +func (f *noEmbeddingProxyService) RegisterToken() string { return "" } +func (f *noEmbeddingProxyService) Invalidate() {} +func (f *noEmbeddingProxyService) RevokeToken() {} + +func (f *noEmbeddingProxyService) ClickHouseDatasources() []string { return nil } +func (f *noEmbeddingProxyService) ClickHouseDatasourceInfo() []types.DatasourceInfo { return nil } + +func (f *noEmbeddingProxyService) ClickHouseQuery(context.Context, string, string, url.Values) ([]byte, error) { + return nil, nil +} + +func (f *noEmbeddingProxyService) PrometheusDatasourceInfo() []types.DatasourceInfo { return nil } +func (f *noEmbeddingProxyService) LokiDatasourceInfo() []types.DatasourceInfo { return nil } +func (f *noEmbeddingProxyService) BenchmarkoorDatasourceInfo() []types.DatasourceInfo { return nil } +func (f *noEmbeddingProxyService) ComputeDatasourceInfo() []types.DatasourceInfo { return nil } +func (f *noEmbeddingProxyService) EthNodeAvailable() bool { return false } +func (f *noEmbeddingProxyService) EthNodeDatasourceInfo() []types.DatasourceInfo { return nil } +func (f *noEmbeddingProxyService) EmbeddingAvailable() bool { return false } +func (f *noEmbeddingProxyService) EmbeddingModel() string { return "" } + +// TestOnDiscoverNeverRacesClose hammers concurrent OnDiscover and Close calls +// across many trials. Before the fix, a discovery goroutine calling +// wg.Add(1) could land while Close's wg.Wait() was already unblocking, +// which the sync.WaitGroup docs call misuse and Go's runtime turns into a +// panic. This must run with -race for the assertion to mean anything. +func TestOnDiscoverNeverRacesClose(t *testing.T) { + log := logrus.New() + log.SetLevel(logrus.PanicLevel) + + const trials = 200 + + for i := 0; i < trials; i++ { + proxySvc := newNoEmbeddingProxyService(t) + r := &Runtime{ + log: log, + proxyService: proxySvc, + stop: make(chan struct{}), + } + + var wg sync.WaitGroup + wg.Add(2) + + go func() { + defer wg.Done() + + for j := 0; j < 5; j++ { + r.OnDiscover() + } + }() + + go func() { + defer wg.Done() + + time.Sleep(time.Millisecond) + require.NoError(t, r.Close()) + }() + + wg.Wait() + } +} diff --git a/pkg/searchruntime/runtime.go b/pkg/searchruntime/runtime.go index 5828c560..04a85208 100644 --- a/pkg/searchruntime/runtime.go +++ b/pkg/searchruntime/runtime.go @@ -55,6 +55,14 @@ type Runtime struct { stop chan struct{} wg sync.WaitGroup + + // closeMu pairs a decision to wg.Add (in OnDiscover) with the decision to + // start closing (in Close) so the two can never race: OnDiscover checks + // closed and adds to wg under the same lock Close takes to set closed + // before it calls wg.Wait. That ordering rules out wg.Add ever landing + // concurrently with wg.Wait, which the WaitGroup docs call misuse. + closeMu sync.Mutex + closed bool } // exampleRefreshInterval is how often the example search index is rebuilt to @@ -201,7 +209,16 @@ func (r *Runtime) OnDiscover() { return } + r.closeMu.Lock() + if r.closed { + r.closeMu.Unlock() + r.activating.Store(false) + + return + } + r.wg.Add(1) + r.closeMu.Unlock() go func() { defer r.wg.Done() @@ -255,12 +272,18 @@ func (r *Runtime) fetchExternalRegistries(ctx context.Context, specsCfg config.C wg.Wait() } -// Close stops the background refresher and releases the shared embedder. +// Close stops the background refresher and releases the shared embedder. Once +// Close has started, OnDiscover becomes a permanent no-op (see closeMu), so no +// caller still invoking it from a discovery goroutine that hasn't noticed +// shutdown yet can add to wg after this point. func (r *Runtime) Close() error { if r == nil { return nil } + r.closeMu.Lock() + r.closed = true + if r.stop != nil { select { case <-r.stop: @@ -268,6 +291,7 @@ func (r *Runtime) Close() error { close(r.stop) } } + r.closeMu.Unlock() r.wg.Wait() diff --git a/pkg/server/builder.go b/pkg/server/builder.go index 3469ae98..76a450cb 100644 --- a/pkg/server/builder.go +++ b/pkg/server/builder.go @@ -153,11 +153,16 @@ func (b *Builder) Build(ctx context.Context) (Service, error) { cleanup := func(stopCtx context.Context) error { var errs []error - if err := searchRuntime.Close(); err != nil { + // Stop the app (and with it, every proxy client's background discovery + // goroutine) before closing the search runtime. application.Stop joins + // those goroutines rather than merely signaling them, so once it + // returns no discovery tick can still be in flight to call OnDiscover + // while searchRuntime.Close is waiting for its own goroutines to drain. + if err := application.Stop(stopCtx); err != nil { errs = append(errs, err) } - if err := application.Stop(stopCtx); err != nil { + if err := searchRuntime.Close(); err != nil { errs = append(errs, err) }