feat(router): Add wait_for_hook for Cosmo Streams#3099
Conversation
|
Preview deployment for your docs. Learn more about Mintlify Previews.
💡 Tip: Enable Workflows to automatically generate PRs for you. |
|
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 ignored due to path filters (2)
📒 Files selected for processing (2)
🚧 Files skipped from review as they are similar to previous changes (2)
WalkthroughAdds a ChangesReceive hook configuration and execution
Estimated code review effort: 3 (Moderate) | ~20 minutes 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches⚔️ Resolve merge conflicts
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 |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 5
🤖 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`:
- Line 2015: Rewrite the description for
handlers.on_receive_events.wait_for_hooks as several short, declarative
sentences. Preserve all existing details about ordered completion, concurrent
subscriber handling, max_concurrent_handlers, handler_timeout, and possible
out-of-order delivery.
In `@router/pkg/config/config.go`:
- Line 877: Set the shared wait_for_hooks default to true by updating the Go
configuration declaration in router/pkg/config/config.go:877, then align the
schema default in router/pkg/config/config.schema.json:3291-3294, the generated
defaults fixture in router/pkg/config/testdata/config_defaults.json:409, the
explicit full-config fixture in router/pkg/config/testdata/config_full.json:835,
and the documented default in docs-website/router/configuration.mdx:2015. Keep
all configuration surfaces consistent with the finalized runtime contract.
In `@router/pkg/pubsub/datasource/subscription_event_updater.go`:
- Around line 51-61: Update the ordered hook execution loop around subscription
event processing to pass the timeout-controlled context instead of
context.Background(), so handler_timeout can cancel blocked hooks. Add the same
panic recovery/isolation behavior used by updateSubscription around each hook
invocation, ensuring one panicking hook does not abort ordered event processing
while preserving existing error handling.
- Around line 55-62: Update the error-handling path in the subscription event
updater so an ordered hook-chain failure closes the affected subscription before
continuing. Match the concurrent path’s cleanup behavior by invoking the
existing subscription-close mechanism in the err != nil branch, rather than
silently dropping the batch and leaving the subscription open.
- Around line 83-91: Update the ordered rounds-building loop to check whether
subEvents[r] is nil before calling GetData. Skip nil hook results while
preserving the existing subData population for non-nil events, matching the
concurrent path’s behavior.
🪄 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: 85d82c86-2379-4479-b237-1094e7fec81a
⛔ Files ignored due to path filters (2)
router-tests/go.sumis excluded by!**/*.sumrouter/go.sumis excluded by!**/*.sum
📒 Files selected for processing (12)
docs-website/router/configuration.mdxrouter-tests/go.modrouter/core/factoryresolver.gorouter/core/router.gorouter/core/router_config.gorouter/go.modrouter/pkg/config/config.gorouter/pkg/config/config.schema.jsonrouter/pkg/config/testdata/config_defaults.jsonrouter/pkg/config/testdata/config_full.jsonrouter/pkg/pubsub/datasource/hooks.gorouter/pkg/pubsub/datasource/subscription_event_updater.go
| | :------------------- | :------------------------------------------------- | :--------------------- | :------------------------------------------------------------------------------------- | :- | | ||
| | | handlers.on_receive_events.handler_timeout | <Icon icon="square" /> | The maximum time to wait for all OnReceiveEvents handlers to complete per event-batch. | 5s | | ||
| | | handlers.on_receive_events.max_concurrent_handlers | <Icon icon="square" /> | The maximum number of concurrent OnReceiveEvents handlers per trigger. | 100 | | ||
| | | handlers.on_receive_events.wait_for_hooks | <Icon icon="square" /> | If true, all OnReceiveEvents handlers for all subscribers run and complete for one event before the router moves on to the next event, so subscribers receive events atomically and in order. If false, handlers for each subscriber run concurrently and independently (subject to `max_concurrent_handlers` and `handler_timeout`), which may deliver events out of order when handlers are slow or time out. | true | |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Split this description into short declarative sentences.
The current table cell packs ordered behavior, concurrent behavior, limits, and timeout consequences into one sentence.
As per coding guidelines, “Prefer short, declarative sentences. If a sentence has more than one comma-separated clause, consider splitting it.”
🤖 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 `@docs-website/router/configuration.mdx` at line 2015, Rewrite the description
for handlers.on_receive_events.wait_for_hooks as several short, declarative
sentences. Preserve all existing details about ordered completion, concurrent
subscriber handling, max_concurrent_handlers, handler_timeout, and possible
out-of-order delivery.
Source: Coding guidelines
| type OnReceiveEventsConfiguration struct { | ||
| MaxConcurrentHandlers int `yaml:"max_concurrent_handlers" envDefault:"100"` | ||
| HandlerTimeout time.Duration `yaml:"handler_timeout" envDefault:"5s"` | ||
| WaitForHooks bool `yaml:"wait_for_hooks" envDefault:"false"` |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Make the wait_for_hooks default consistent across all configuration surfaces.
The Go declaration defaults to false, while every other changed surface uses true. Prefer updating the Go envDefault to true if the fixtures and public documentation represent the intended behavior.
router/pkg/config/config.go#L877-L877: changeenvDefaultto the intended shared default.router/pkg/config/config.schema.json#L3291-L3294: keep the schema default aligned with the Go loader.router/pkg/config/testdata/config_defaults.json#L409-L409: regenerate or update after selecting the runtime default.router/pkg/config/testdata/config_full.json#L835-L835: use the same explicit value as the finalized contract.docs-website/router/configuration.mdx#L2015-L2015: document the finalized default.
📍 Affects 5 files
router/pkg/config/config.go#L877-L877(this comment)router/pkg/config/config.schema.json#L3291-L3294router/pkg/config/testdata/config_defaults.json#L409-L409router/pkg/config/testdata/config_full.json#L835-L835docs-website/router/configuration.mdx#L2015-L2015
🤖 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/pkg/config/config.go` at line 877, Set the shared wait_for_hooks
default to true by updating the Go configuration declaration in
router/pkg/config/config.go:877, then align the schema default in
router/pkg/config/config.schema.json:3291-3294, the generated defaults fixture
in router/pkg/config/testdata/config_defaults.json:409, the explicit full-config
fixture in router/pkg/config/testdata/config_full.json:835, and the documented
default in docs-website/router/configuration.mdx:2015. Keep all configuration
surfaces consistent with the finalized runtime contract.
There was a problem hiding this comment.
Will be tackled eventually
There was a problem hiding this comment.
@dkorittki Understood. Would you like me to create a follow-up GitHub issue to track aligning the wait_for_hooks default across the runtime config, schema, fixtures, and documentation?
| var err error | ||
| for i := range hooks { | ||
| // TODO: replace context.Background() with something proper | ||
| subEvents, err = hooks[i](subCtx, context.Background(), s.subscriptionEventConfiguration, s.eventBuilder, subEvents) | ||
| if err != nil { | ||
| break | ||
| } | ||
| } | ||
| if err != nil { | ||
| // TODO: decide whether to CloseSubscription here or swallow the error | ||
| continue |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Preserve timeout and panic isolation in ordered mode.
Line 54 uses context.Background(), so handler_timeout cannot cancel a blocked hook. This synchronous path also lacks the panic recovery used by updateSubscription, allowing one custom hook to abort event processing.
Also applies to: 97-103
🤖 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/pkg/pubsub/datasource/subscription_event_updater.go` around lines 51 -
61, Update the ordered hook execution loop around subscription event processing
to pass the timeout-controlled context instead of context.Background(), so
handler_timeout can cancel blocked hooks. Add the same panic recovery/isolation
behavior used by updateSubscription around each hook invocation, ensuring one
panicking hook does not abort ordered event processing while preserving existing
error handling.
There was a problem hiding this comment.
This will be fixed the upcoming days
There was a problem hiding this comment.
@dkorittki, thanks for the update. I’ll keep this review thread open until the timeout propagation and panic isolation are implemented.
dkorittki
changed the title
wait_for_hook option for Cosmo Streams' OnReceiveEvent hookwait_for_hook for Cosmo Streams OnReceiveEvent
dkorittki
changed the title
wait_for_hook for Cosmo Streams OnReceiveEventwait_for_hook for Cosmo Streams
1 participant
Currently, when using Cosmo Streams with the
OnReceiveEventhook the router performs all hook processing for all subscribers on a trigger concurrently. Once any hook instance finished processing a subscriber it immediately asks the engine to perform response resolving and sending the message to clients. This can happen concurrently for any subscriber.This concurrent calls cause mutex stalls if the router has lots of subscribers. Since the engine does not know when the router wants to update the another subscriber it has to use mutexes synchronize the subscription updater.
We can avoid this by giving the engine a bulk list of what to update, instead of having it to guess when to update what.
In this PR we add a new option called
events.handlers.on_receive_events.wait_for_hooksto do just that.If set to
truethe router first processes theOnReceiveEventhook for every subscriber this event is for.Only after that do we call the engine and handing over a list of all subscribers with their event payload.
The engine can serially work on this, avoiding a lot of mutex work in the process.
The default of this option is
false. We can think about making ittrue.This has a bunch of implications:
Note: None of the hook processing is async for now. It's one goroutine doing it all. It's debateble wether spawning lots of small routines and having the runtime do the housekeeping is more efficient than simply letting it run synchronously.
Corresponding engine PR: wundergraph/graphql-go-tools#1603
Todos
This might grow.
Summary by CodeRabbit
wait_for_hooksoption for subscription receive-event handlers.Checklist
Open Source AI Manifesto
This project follows the principles of the Open Source AI Manifesto. Please ensure your contribution aligns with its principles.