fix(router): fix pubsub goroutine leak - #3047
Conversation
Router-nonroot image scan passed✅ No security vulnerabilities found in image: |
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (1)
🚧 Files skipped from review as they are similar to previous changes (1)
WalkthroughPub/sub provider lifecycle moves from graphServer to graphMux, so mux construction starts providers and mux shutdown stops them. A shutdown reuse test is added, and NATS event log assertions now expect two lifecycle entries per provider. ChangesPub/sub provider lifecycle relocation
Estimated code review effort: 3 (Moderate) | ~25 minutes Possibly related PRs
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
Warning There were issues while running some tools. Please review the errors and either fix the tool's configuration or disable the tool if it's a critical failure. 🔧 golangci-lint (2.12.2)level=error msg="[linters_context] typechecking error: pattern ./...: directory prefix . does not contain main module or its selected dependencies" Comment |
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #3047 +/- ##
===========================================
+ Coverage 48.40% 61.44% +13.04%
===========================================
Files 1128 261 -867
Lines 155971 30596 -125375
Branches 10684 0 -10684
===========================================
- Hits 75499 18800 -56699
+ Misses 78618 10280 -68338
+ Partials 1854 1516 -338
🚀 New features to boost your workflow:
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
router/core/graph_server.go (1)
2314-2338: 🩺 Stability & Availability | 🔴 Critical | ⚡ Quick winShared
time.Timerchannel can deadlockprovidersActionWithTimeoutwhen multiple providers time out concurrently.
timer.Conly ever delivers a single value. When N providers race against the same timer and the timeout fires, only one of the N goroutines'selectstatements can receive fromtimer.C; the remaining N-1 goroutines block indefinitely on theirselect(sincecancellableCtxis never canceled beforeprovidersGroup.Wait()returns, and their underlyingaction(cancellableCtx, provider)goroutine has no other exit signal). This deadlocksprovidersGroup.Wait()and leaks goroutines — the very failure mode this PR is meant to fix — wheneverstartPubsubProviders/stopPubsubProvidershandles multiple providers that time out together (e.g., during shutdown with several hung pub/sub connections).Use a cancellable timeout context instead of a shared timer so
Done()broadcasts to all waiters, and propagate it intoactionso providers can actually observe the timeout:🔧 Proposed fix
func providersActionWithTimeout(ctx context.Context, providers []datasource.Provider, action func(ctx context.Context, provider datasource.Provider) error, timeout time.Duration, timeoutMessage string) error { - cancellableCtx, cancel := context.WithCancel(ctx) + timeoutCtx, cancel := context.WithTimeout(ctx, timeout) defer cancel() - timer := time.NewTimer(timeout) - defer timer.Stop() - providersGroup := new(errgroup.Group) for _, provider := range providers { + provider := provider providersGroup.Go(func() error { actionDone := make(chan error, 1) go func() { - actionDone <- action(cancellableCtx, provider) + actionDone <- action(timeoutCtx, provider) }() select { case err := <-actionDone: return err - case <-timer.C: + case <-timeoutCtx.Done(): return errors.New(timeoutMessage) } }) } return providersGroup.Wait() }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@router/core/graph_server.go` around lines 2314 - 2338, providersActionWithTimeout is using a single shared time.Timer, so only one goroutine can observe timer.C and the rest may block forever when several providers time out together. Replace the timer/cancel setup with a timeout-derived context in providersActionWithTimeout so Done() broadcasts to every waiter, and pass that context through to action and the provider goroutines. Keep the existing errgroup wait flow, but make sure each provider select listens on the shared context cancellation rather than a shared timer channel.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@router/core/graph_server.go`:
- Around line 2314-2338: providersActionWithTimeout is using a single shared
time.Timer, so only one goroutine can observe timer.C and the rest may block
forever when several providers time out together. Replace the timer/cancel setup
with a timeout-derived context in providersActionWithTimeout so Done()
broadcasts to every waiter, and pass that context through to action and the
provider goroutines. Keep the existing errgroup wait flow, but make sure each
provider select listens on the shared context cancellation rather than a shared
timer channel.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: a30d51c8-3f34-4a2d-a851-3c29bbec4b2c
📒 Files selected for processing (3)
router-tests/events/nats_events_test.gorouter/core/graph_server.gorouter/core/graph_server_test.go
|
897c71f fixes what coderrabbit found in #3047 (review) |
3 participants
Based on @endigma 's findings and fix in #3038. I took over to polish and fix tests.
Pubsub providers were stored on the graph server. The reality is that every graph mux has their own provider instances. This lead to the graph servers starting providers for each graph mux they spawn but only stopping the ones from the latest graph mux. This leaked provider goroutines on router hot reloads. It only happens with active feature flags and only if Cosmo Streams is active.
The order is as follow:
Situation: one base graph mux with a provider, one feature flag graph mux with a provider
pubSubProvidersfieldpubSubProvidersfield (overwriting the ones from the prior mux)This pull request fixes it by
Summary by CodeRabbit
Summary by CodeRabbit
Bug Fixes
Tests
TestGraphServerShutdownto confirm providers are shut down or retained based on mux reuse.Checklist
Open Source AI Manifesto
This project follows the principles of the Open Source AI Manifesto. Please ensure your contribution aligns with its principles.