feat: router start more lenient on cosmo streams problems#2994
Open
alepane21 wants to merge 14 commits into
Open
feat: router start more lenient on cosmo streams problems#2994alepane21 wants to merge 14 commits into
alepane21 wants to merge 14 commits into
Conversation
Contributor
|
Preview deployment for your docs. Learn more about Mintlify Previews.
💡 Tip: Enable Workflows to automatically generate PRs for you. |
Contributor
|
Important Review skippedThis PR was authored by the user configured for CodeRabbit reviews. CodeRabbit does not review PRs authored by this user. It's recommended to use a dedicated user account to post CodeRabbit review feedback. ⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
Comment |
|
❌ Internal Query Planner CI failed to run. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 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.
Inline comments:
In `@docs-website/router/configuration.mdx`:
- Around line 1849-1851: The description for the skip_missing_providers
configuration parameter is currently a single long sentence that combines
multiple distinct behaviors, reducing readability. Replace this long description
with multiple shorter declarative sentences or bullet points that separately
convey each behavior: allowing the router to start when a provider is missing
from the execution config, logging an error when this occurs, disabling only the
affected fields instead of failing startup, and noting that by default a missing
provider prevents startup. Follow documentation guidelines by breaking complex
ideas into concise, focused statements for better reference-doc usability.
In `@docs-website/router/cosmo-streams.mdx`:
- Around line 193-194: Convert the lenient-mode behavior paragraph (in the
section describing the skip_missing_providers configuration) into a structured
bulleted or numbered list. Break apart the current paragraph that describes the
router's behavior with the skip_missing_providers setting into separate,
discrete list items where each item focuses on one specific effect or action:
how to enable the feature, what the router does in response, what continues to
work, what fails, and what operational steps should be taken. Use short,
declarative sentences for each list item to improve readability and scannability
of this reference documentation.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 89ee15bb-27d0-47bf-a88a-23ad6866dcaf
📒 Files selected for processing (10)
docs-website/router/configuration.mdxdocs-website/router/cosmo-streams.mdxrouter/pkg/config/config.gorouter/pkg/config/config.schema.jsonrouter/pkg/config/config_test.gorouter/pkg/config/fixtures/full.yamlrouter/pkg/config/testdata/config_defaults.jsonrouter/pkg/config/testdata/config_full.jsonrouter/pkg/pubsub/pubsub.gorouter/pkg/pubsub/pubsub_test.go
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #2994 +/- ##
===========================================
- Coverage 61.77% 39.37% -22.41%
===========================================
Files 261 14 -247
Lines 30612 1651 -28961
===========================================
- Hits 18912 650 -18262
+ Misses 10184 873 -9311
+ Partials 1516 128 -1388 🚀 New features to boost your workflow:
|
…ils-in-production-due-to
…r-startup-fails-in-production-due-to' into ale/eng-9792-investigation-router-startup-fails-in-production-due-to
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 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.
Inline comments:
In `@router/core/graph_server.go`:
- Around line 2261-2277: The adapter implementations for NATS, Kafka, and Redis
do not respect context cancellation in their Startup methods, which allows the
Startup goroutine to continue executing and establish a connection even after
the provider is marked unavailable via MarkUnavailable(). When this happens, the
Shutdown method skips cleanup because the provider is already marked
unavailable, leaving an orphaned connection. Fix this by ensuring that all
adapter Startup implementations properly check and respect context cancellation,
allowing in-flight Startup operations to be interrupted when a provider is
marked unavailable, preventing connections from being established for
unavailable providers.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: f389e8e6-2dc3-43ed-bfdc-27446df21328
📒 Files selected for processing (19)
docs-website/router/configuration.mdxdocs-website/router/cosmo-streams.mdxrouter/core/graph_server.gorouter/core/graph_server_test.gorouter/demo.config.yamlrouter/pkg/config/config.gorouter/pkg/config/config.schema.jsonrouter/pkg/config/config_test.gorouter/pkg/config/fixtures/full.yamlrouter/pkg/config/testdata/config_defaults.jsonrouter/pkg/config/testdata/config_full.jsonrouter/pkg/pubsub/datasource/pubsubprovider.gorouter/pkg/pubsub/datasource/pubsubprovider_test.gorouter/pkg/pubsub/nats/engine_datasource.gorouter/pkg/pubsub/nats/engine_datasource_test.gorouter/pkg/pubsub/pubsub.gorouter/pkg/pubsub/pubsub_test.gorouter/pkg/pubsub/redis/adapter.gorouter/pkg/pubsub/redis/adapter_test.go
✅ Files skipped from review due to trivial changes (2)
- router/pkg/config/testdata/config_defaults.json
- docs-website/router/configuration.mdx
🚧 Files skipped from review as they are similar to previous changes (2)
- router/pkg/pubsub/pubsub.go
- router/pkg/pubsub/pubsub_test.go
alepane21
changed the title
…ils-in-production-due-to
…ation-router-startup-fails-in-production-due-to
Contributor
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 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.
Inline comments:
In `@router-tests/events/kafka_skip_unavailable_test.go`:
- Around line 142-145: The goroutine in the kafka skip-unavailable test is
sending the result of client.Run() into an unbuffered channel that is never
read, which can block forever during cleanup. Update the test around clientRunCh
and the client.Run() goroutine so the result is either drained or sent to a
buffered channel, and make sure the goroutine can exit cleanly after
client.Close() without blocking.
In `@router-tests/events/redis_skip_unavailable_test.go`:
- Around line 139-142: The goroutine in the redis skip-unavailable test can hang
because client.Run() sends to an unbuffered runCh that is never read. Update the
test around the client.Run() launch to avoid blocking on the send, either by
making runCh buffered or by consuming the result with a select/receive during
teardown, so the goroutine can exit cleanly. Use the client.Run and runCh
symbols to locate the test block.
In `@router-tests/testenv/tcpproxy.go`:
- Around line 71-85: SetReachable(false) only closes the current snapshot of
p.conns, so a connection accepted just before the toggle can still be tracked
afterward and continue forwarding. Update ToggleableProxy’s connection
registration path in trackConn (and any related accept/registration logic) to
reject or immediately close connections when p.reachable is false, so late
arrivals cannot be added after SetReachable(false) has been called. Keep the
behavior centered around SetReachable and trackConn so the proxy fully simulates
broker unavailability.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 797805ee-72b0-4480-b549-f9638a045e4f
📒 Files selected for processing (20)
docs-website/router/configuration.mdxdocs-website/router/cosmo-streams.mdxrouter-tests/events/kafka_skip_unavailable_test.gorouter-tests/events/nats_skip_unavailable_test.gorouter-tests/events/redis_skip_unavailable_test.gorouter-tests/testenv/tcpproxy.gorouter/core/graph_server.gorouter/core/graph_server_test.gorouter/core/router.gorouter/internal/rediscloser/rediscloser.gorouter/internal/rediscloser/rediscloser_test.gorouter/pkg/config/config.gorouter/pkg/pubsub/datasource/provider.gorouter/pkg/pubsub/datasource/pubsubprovider.gorouter/pkg/pubsub/kafka/adapter.gorouter/pkg/pubsub/nats/adapter.gorouter/pkg/pubsub/nats/provider_builder.gorouter/pkg/pubsub/nats/provider_builder_test.gorouter/pkg/pubsub/pubsub.gorouter/pkg/pubsub/redis/adapter.go
✅ Files skipped from review due to trivial changes (4)
- router/core/router.go
- docs-website/router/configuration.mdx
- router/pkg/pubsub/datasource/pubsubprovider.go
- docs-website/router/cosmo-streams.mdx
🚧 Files skipped from review as they are similar to previous changes (3)
- router/pkg/config/config.go
- router/core/graph_server.go
- router/pkg/pubsub/pubsub.go
Contributor
There was a problem hiding this comment.
🧹 Nitpick comments (1)
router-tests/testenv/tcpproxy_test.go (1)
76-81: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winAssert that
SetReachable(false)drops an already-established proxied connection.This block only proves future dials fail. It does not exercise the new
ToggleableProxy.SetReachable(false)contract that closes currently forwarded connections, so a regression in that path would still pass here.Suggested test extension
// Once reachable, traffic is forwarded to the upstream and echoed back. proxy.SetReachable(true) require.Eventually(t, func() bool { got, rtErr := proxyRoundTrip(proxy.Addr()) return rtErr == nil && got == "ping" }, 5*time.Second, 50*time.Millisecond) + // Keep one proxied connection open so we can verify it gets torn down. + conn, err := net.DialTimeout("tcp", proxy.Addr(), 2*time.Second) + require.NoError(t, err) + t.Cleanup(func() { _ = conn.Close() }) + require.NoError(t, conn.SetDeadline(time.Now().Add(2*time.Second))) + _, err = conn.Write([]byte("ping")) + require.NoError(t, err) + buf := make([]byte, 4) + _, err = io.ReadFull(conn, buf) + require.NoError(t, err) + require.Equal(t, "ping", string(buf)) + // Toggled back to unreachable: new connections must not forward anymore. proxy.SetReachable(false) + require.Eventually(t, func() bool { + _, writeErr := conn.Write([]byte("ping")) + if writeErr != nil { + return true + } + _, readErr := io.ReadFull(conn, buf) + return readErr != nil + }, 5*time.Second, 50*time.Millisecond) require.Eventually(t, func() bool { _, rtErr := proxyRoundTrip(proxy.Addr()) return rtErr != nil }, 5*time.Second, 50*time.Millisecond)🤖 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-tests/testenv/tcpproxy_test.go` around lines 76 - 81, Update the `ToggleableProxy` test to verify that `SetReachable(false)` also closes an already-established proxied connection, not just future dials. In `tcpproxy_test.go`, extend the existing `proxyRoundTrip`/`proxy.SetReachable(false)` flow by keeping one proxied connection open before toggling, then assert that the existing connection is dropped after the reachability change while new `proxyRoundTrip(proxy.Addr())` calls still fail. Use the `ToggleableProxy`, `SetReachable`, and `proxyRoundTrip` symbols to locate and extend the current test case.
🤖 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.
Nitpick comments:
In `@router-tests/testenv/tcpproxy_test.go`:
- Around line 76-81: Update the `ToggleableProxy` test to verify that
`SetReachable(false)` also closes an already-established proxied connection, not
just future dials. In `tcpproxy_test.go`, extend the existing
`proxyRoundTrip`/`proxy.SetReachable(false)` flow by keeping one proxied
connection open before toggling, then assert that the existing connection is
dropped after the reachability change while new `proxyRoundTrip(proxy.Addr())`
calls still fail. Use the `ToggleableProxy`, `SetReachable`, and
`proxyRoundTrip` symbols to locate and extend the current test case.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 619b94f7-478e-41f3-921c-2101e72b6377
📒 Files selected for processing (2)
router-tests/testenv/tcpproxy.gorouter-tests/testenv/tcpproxy_test.go
🚧 Files skipped from review as they are similar to previous changes (1)
- router-tests/testenv/tcpproxy.go
…ation-router-startup-fails-in-production-due-to
dkorittki
requested changes
Jul 9, 2026
| // is used. Callers that must guarantee NewRedisCloser returns within a deadline (e.g. a | ||
| // bounded provider startup) should pass a context with a timeout so a black-holed broker | ||
| // cannot block the ping for the full go-redis dial timeout. | ||
| Context context.Context |
Contributor
There was a problem hiding this comment.
Instead of having this here I would put it as the first parameter on NewRedisCloser
dkorittki
reviewed
Jul 9, 2026
| } | ||
|
|
||
| if config.SkipUnavailableProviders { | ||
| logger.Warn("EDFS lenient mode is enabled (events.skip_unavailable_providers=true): the router will start even if an event provider referenced by the execution config is undefined or unreachable, disabling only the affected fields") |
Contributor
There was a problem hiding this comment.
I don't think the warning is necessary. When the router can't connect to the provider we log errors anyway. If we can connect there is no problem so we don't need this warning, too.
dkorittki
reviewed
Jul 10, 2026
Comment on lines
+2313
to
+2345
| // Each provider is bounded by its own timer. A single shared timer would only | ||
| // ever deliver its fire to one goroutine, so multiple slow/unreachable providers | ||
| // would block the others forever and hang startup. | ||
| timer := time.NewTimer(timeout) | ||
| defer timer.Stop() | ||
|
|
||
| actionDone := make(chan error, 1) | ||
| go func() { | ||
| actionDone <- action(timeoutCtx, provider) | ||
| }() | ||
| var actionErr error | ||
| select { | ||
| case err := <-actionDone: | ||
| return err | ||
| case <-timeoutCtx.Done(): | ||
| return errors.New(timeoutMessage) | ||
| } | ||
| if actionErr == nil || !continueOnError { | ||
| return actionErr | ||
| } | ||
| // Lenient mode (events.skip_unavailable_providers): the provider failed to start. | ||
| // Log a distinct error and let the router keep running with the affected fields | ||
| // unavailable. When the cause is an unreachable broker the adapter retains a | ||
| // resilient client that reconnects in the background, so those fields recover | ||
| // without a restart once the broker is reachable. When the cause is a | ||
| // configuration error (e.g. an invalid URL) the provider cannot recover until the | ||
| // configuration is fixed; the underlying cause is attached as the error field. | ||
| s.logger.Error("EDFS provider could not be started at startup; the router will keep running and the fields backed by this provider are temporarily unavailable. An unreachable broker reconnects and recovers automatically without a restart; see the error for the cause", | ||
| zap.String("provider_id", provider.ID()), | ||
| zap.String("provider_type", provider.TypeID()), | ||
| zap.Error(actionErr), | ||
| ) | ||
| return nil |
Contributor
There was a problem hiding this comment.
The lates pull from main broke your code. The timer is not needed anymore, got replaced by timeoutCtx. Here's a suggestion from me. After that this method needs to be made again a method of graphServer and continueOnError parameter needs to be readded to method signature.
Suggested change
| // Each provider is bounded by its own timer. A single shared timer would only | |
| // ever deliver its fire to one goroutine, so multiple slow/unreachable providers | |
| // would block the others forever and hang startup. | |
| timer := time.NewTimer(timeout) | |
| defer timer.Stop() | |
| actionDone := make(chan error, 1) | |
| go func() { | |
| actionDone <- action(timeoutCtx, provider) | |
| }() | |
| var actionErr error | |
| select { | |
| case err := <-actionDone: | |
| return err | |
| case <-timeoutCtx.Done(): | |
| return errors.New(timeoutMessage) | |
| } | |
| if actionErr == nil || !continueOnError { | |
| return actionErr | |
| } | |
| // Lenient mode (events.skip_unavailable_providers): the provider failed to start. | |
| // Log a distinct error and let the router keep running with the affected fields | |
| // unavailable. When the cause is an unreachable broker the adapter retains a | |
| // resilient client that reconnects in the background, so those fields recover | |
| // without a restart once the broker is reachable. When the cause is a | |
| // configuration error (e.g. an invalid URL) the provider cannot recover until the | |
| // configuration is fixed; the underlying cause is attached as the error field. | |
| s.logger.Error("EDFS provider could not be started at startup; the router will keep running and the fields backed by this provider are temporarily unavailable. An unreachable broker reconnects and recovers automatically without a restart; see the error for the cause", | |
| zap.String("provider_id", provider.ID()), | |
| zap.String("provider_type", provider.TypeID()), | |
| zap.Error(actionErr), | |
| ) | |
| return nil | |
| actionDone := make(chan error, 1) | |
| go func() { | |
| actionDone <- action(timeoutCtx, provider) | |
| }() | |
| var err error | |
| select { | |
| case err = <-actionDone: | |
| case <-timeoutCtx.Done(): | |
| err = errors.New(timeoutMessage) | |
| } | |
| if err == nil || !continueOnError { | |
| return err | |
| } | |
| s.logger.Error("EDFS provider could not be started at startup; the router will keep running and the fields backed by this provider are temporarily unavailable. An unreachable broker reconnects and recovers automatically without a restart; see the error for the cause", | |
| zap.String("provider_id", provider.ID()), | |
| zap.String("provider_type", provider.TypeID()), | |
| zap.Error(err), | |
| ) | |
| return nil |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Introduce a way to start the router even if Cosmo Streams provider are not available or not responding.
If the provider is not defined, the router will start, log out an error, and that's it.
If the provider is defined but not reachable, it will start, log the error and try to connect to the provider, without the router even needing to reload.
To test the behaviour I decided to create a small tcp proxy adapter (almost fully AI generated) to simulate those condition. The other way would have been to introduce something like toxiproxy, but I think it would have been an overkill at the moment.
By default I left the current behaviour, only if the new mode is enabled (with config or env), it will be enabled.
Summary by CodeRabbit
New Features
Bug Fixes
Checklist
Open Source AI Manifesto
This project follows the principles of the Open Source AI Manifesto. Please ensure your contribution aligns with its principles.