fix(router): fix inefficient trigger id generation - #2950
Merged
dkorittki merged 15 commits intoJun 15, 2026
Merged
Conversation
Router image scan passed✅ No security vulnerabilities found in image: |
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #2950 +/- ##
==========================================
+ Coverage 65.42% 66.48% +1.06%
==========================================
Files 275 258 -17
Lines 29014 27559 -1455
==========================================
- Hits 18981 18323 -658
+ Misses 8530 7779 -751
+ Partials 1503 1457 -46
🚀 New features to boost your workflow:
|
Contributor
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
WalkthroughReplace subscription uniqueRequestID with triggerHashInput, update Kafka/NATS/Redis factories to use it, add NATS Flush after channel subscriptions, introduce EDFS+NATS integration tests for trigger deduplication, add a test helper to assert trigger counts, and bump graphql-go-tools module versions. ChangesTrigger Deduplication and Hash-Input Refactoring
🎯 3 (Moderate) | ⏱️ ~25 minutes 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
🧹 Nitpick comments (7)
router-tests/events/trigger_test.go (7)
265-268: 💤 Low valueRemove redundant manual deadline before
testenv.WSReadJSON.Same as previous instances: the manual
SetReadDeadlineis redundant when usingtestenv.WSReadJSON.As per coding guidelines: "Use manual SetReadDeadline with conn.ReadJSON only when expecting errors (e.g., websocket close after config hot reload); otherwise use testenv.WSReadJSON"
🤖 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/events/trigger_test.go` around lines 265 - 268, Remove the redundant manual deadline before calling testenv.WSReadJSON: delete the conn.SetReadDeadline(time.Now().Add(time.Second)) call and its require.NoError(t, err) check so the code directly calls testenv.WSReadJSON(t, conn, &complete); this keeps behavior consistent with other tests that rely on testenv.WSReadJSON rather than manual SetReadDeadline/conn.ReadJSON.Source: Coding guidelines
234-237: 💤 Low valueRemove redundant manual deadline before
testenv.WSReadJSON.Same as previous instances: the manual
SetReadDeadlineis redundant when usingtestenv.WSReadJSON.As per coding guidelines: "Use manual SetReadDeadline with conn.ReadJSON only when expecting errors (e.g., websocket close after config hot reload); otherwise use testenv.WSReadJSON"
🤖 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/events/trigger_test.go` around lines 234 - 237, The redundant manual deadline call before the JSON read should be removed: delete the conn.SetReadDeadline(time.Now().Add(time.Second)) and its error check so the test uses testenv.WSReadJSON(t, conn, &complete) directly; locate the calls to conn.SetReadDeadline and require.NoError(t, err) immediately preceding testenv.WSReadJSON in the trigger_test.go test and remove that pair (leave the subsequent testenv.WSReadJSON and require.NoError intact).Source: Coding guidelines
63-66: 💤 Low valueRemove redundant manual deadline before
testenv.WSReadJSON.The coding guideline states to use manual
SetReadDeadlinewithconn.ReadJSONonly when expecting errors. Sincetestenv.WSReadJSONalready sets a 2-second deadline internally with retry logic, the manualSetReadDeadlinehere is redundant and can be removed.🧹 Suggested simplification
err = testenv.WSWriteJSON(t, conn, &testenv.WebSocketMessage{ID: "1", Type: "complete"}) require.NoError(t, err) var complete testenv.WebSocketMessage -err = conn.SetReadDeadline(time.Now().Add(time.Second)) -require.NoError(t, err) err = testenv.WSReadJSON(t, conn, &complete)As per coding guidelines: "Use manual SetReadDeadline with conn.ReadJSON only when expecting errors (e.g., websocket close after config hot reload); otherwise use testenv.WSReadJSON"
🤖 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/events/trigger_test.go` around lines 63 - 66, Remove the redundant manual deadline call before invoking testenv.WSReadJSON: delete the conn.SetReadDeadline(time.Now().Add(time.Second)) and its error check, and keep the subsequent testenv.WSReadJSON(t, conn, &complete) and require.NoError(t, err) as-is; this targets the block using conn.SetReadDeadline, testenv.WSReadJSON, and the complete variable in trigger_test.go so the built-in 2s deadline/retry logic in WSReadJSON is relied upon.Source: Coding guidelines
180-183: 💤 Low valueRemove redundant manual deadline before
testenv.WSReadJSON.Same as previous instances: the manual
SetReadDeadlineis redundant when usingtestenv.WSReadJSON.As per coding guidelines: "Use manual SetReadDeadline with conn.ReadJSON only when expecting errors (e.g., websocket close after config hot reload); otherwise use testenv.WSReadJSON"
🤖 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/events/trigger_test.go` around lines 180 - 183, The manual conn.SetReadDeadline call immediately before calling testenv.WSReadJSON is redundant; remove the SetReadDeadline line so that only testenv.WSReadJSON(t, conn, &complete) remains. Locate the block containing conn.SetReadDeadline, require.NoError(t, err), and testenv.WSReadJSON(t, conn, &complete) and delete the SetReadDeadline call (and its associated require.NoError) so reads rely on testenv.WSReadJSON behavior.Source: Coding guidelines
149-152: 💤 Low valueRemove redundant manual deadline before
testenv.WSReadJSON.Same as previous instances: the manual
SetReadDeadlineis redundant when usingtestenv.WSReadJSON.As per coding guidelines: "Use manual SetReadDeadline with conn.ReadJSON only when expecting errors (e.g., websocket close after config hot reload); otherwise use testenv.WSReadJSON"
🤖 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/events/trigger_test.go` around lines 149 - 152, The manual timeout call before reading is redundant: remove the conn.SetReadDeadline(time.Now().Add(time.Second)) call that immediately precedes testenv.WSReadJSON(t, conn, &complete) in trigger_test.go; keep the testenv.WSReadJSON invocation as-is (and if this removal makes the time import unused, delete that import as well). Ensure any other similar instances use testenv.WSReadJSON without a preceding SetReadDeadline unless the test expects a read error.Source: Coding guidelines
94-97: 💤 Low valueRemove redundant manual deadline before
testenv.WSReadJSON.Same issue as above:
testenv.WSReadJSONhandles deadlines internally, so the manualSetReadDeadlineis unnecessary.🧹 Suggested simplification
err = testenv.WSWriteJSON(t, conn, &testenv.WebSocketMessage{ID: "1", Type: "complete"}) require.NoError(t, err) var complete testenv.WebSocketMessage -err = conn.SetReadDeadline(time.Now().Add(time.Second)) -require.NoError(t, err) err = testenv.WSReadJSON(t, conn, &complete)As per coding guidelines: "Use manual SetReadDeadline with conn.ReadJSON only when expecting errors (e.g., websocket close after config hot reload); otherwise use testenv.WSReadJSON"
🤖 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/events/trigger_test.go` around lines 94 - 97, Remove the manual deadline call before the JSON read: delete the conn.SetReadDeadline(...) and its require.NoError(t, err) that immediately precede testenv.WSReadJSON in trigger_test.go; leave the testenv.WSReadJSON(t, conn, &complete) call as-is because WSReadJSON manages deadlines internally (only use conn.SetReadDeadline together with conn.ReadJSON when you are explicitly testing timeout/error behavior).Source: Coding guidelines
37-37: 💤 Low valueRemove unused parameter from
NATSPublishUntilReceivedcall.The function signature at
testenv.go:2764shows the 4th parameter is explicitly ignored (_ uint64), so passing2here has no effect. Either remove this parameter from all call sites or update the function to use it if a count check is needed.🧹 Suggested cleanup
-xEnv.NATSPublishUntilReceived(xEnv.NatsConnectionDefault, xEnv.GetPubSubName("employeeUpdated.3"), []byte(`{"id":3,"__typename":"Employee"}`), 2, time.Second*10) +xEnv.NATSPublishUntilReceived(xEnv.NatsConnectionDefault, xEnv.GetPubSubName("employeeUpdated.3"), []byte(`{"id":3,"__typename":"Employee"}`), 0, time.Second*10)Or better yet, update the function signature to remove the unused parameter entirely.
🤖 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/events/trigger_test.go` at line 37, Call to xEnv.NATSPublishUntilReceived includes an unused fourth argument (the function signature for NATSPublishUntilReceived declares the fourth parameter as ignored), so remove the extraneous numeric argument from the call site: change xEnv.NATSPublishUntilReceived(xEnv.NatsConnectionDefault, xEnv.GetPubSubName("employeeUpdated.3"), []byte(...), 2, time.Second*10) to call the three meaningful params plus timeout form (omit the ignored uint64), or alternatively modify the NATSPublishUntilReceived function itself to accept and use that count if intended; update all other call sites to match the chosen fix so signatures remain consistent.
🤖 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/events/trigger_test.go`:
- Around line 265-268: Remove the redundant manual deadline before calling
testenv.WSReadJSON: delete the conn.SetReadDeadline(time.Now().Add(time.Second))
call and its require.NoError(t, err) check so the code directly calls
testenv.WSReadJSON(t, conn, &complete); this keeps behavior consistent with
other tests that rely on testenv.WSReadJSON rather than manual
SetReadDeadline/conn.ReadJSON.
- Around line 234-237: The redundant manual deadline call before the JSON read
should be removed: delete the conn.SetReadDeadline(time.Now().Add(time.Second))
and its error check so the test uses testenv.WSReadJSON(t, conn, &complete)
directly; locate the calls to conn.SetReadDeadline and require.NoError(t, err)
immediately preceding testenv.WSReadJSON in the trigger_test.go test and remove
that pair (leave the subsequent testenv.WSReadJSON and require.NoError intact).
- Around line 63-66: Remove the redundant manual deadline call before invoking
testenv.WSReadJSON: delete the conn.SetReadDeadline(time.Now().Add(time.Second))
and its error check, and keep the subsequent testenv.WSReadJSON(t, conn,
&complete) and require.NoError(t, err) as-is; this targets the block using
conn.SetReadDeadline, testenv.WSReadJSON, and the complete variable in
trigger_test.go so the built-in 2s deadline/retry logic in WSReadJSON is relied
upon.
- Around line 180-183: The manual conn.SetReadDeadline call immediately before
calling testenv.WSReadJSON is redundant; remove the SetReadDeadline line so that
only testenv.WSReadJSON(t, conn, &complete) remains. Locate the block containing
conn.SetReadDeadline, require.NoError(t, err), and testenv.WSReadJSON(t, conn,
&complete) and delete the SetReadDeadline call (and its associated
require.NoError) so reads rely on testenv.WSReadJSON behavior.
- Around line 149-152: The manual timeout call before reading is redundant:
remove the conn.SetReadDeadline(time.Now().Add(time.Second)) call that
immediately precedes testenv.WSReadJSON(t, conn, &complete) in trigger_test.go;
keep the testenv.WSReadJSON invocation as-is (and if this removal makes the time
import unused, delete that import as well). Ensure any other similar instances
use testenv.WSReadJSON without a preceding SetReadDeadline unless the test
expects a read error.
- Around line 94-97: Remove the manual deadline call before the JSON read:
delete the conn.SetReadDeadline(...) and its require.NoError(t, err) that
immediately precede testenv.WSReadJSON in trigger_test.go; leave the
testenv.WSReadJSON(t, conn, &complete) call as-is because WSReadJSON manages
deadlines internally (only use conn.SetReadDeadline together with conn.ReadJSON
when you are explicitly testing timeout/error behavior).
- Line 37: Call to xEnv.NATSPublishUntilReceived includes an unused fourth
argument (the function signature for NATSPublishUntilReceived declares the
fourth parameter as ignored), so remove the extraneous numeric argument from the
call site: change xEnv.NATSPublishUntilReceived(xEnv.NatsConnectionDefault,
xEnv.GetPubSubName("employeeUpdated.3"), []byte(...), 2, time.Second*10) to call
the three meaningful params plus timeout form (omit the ignored uint64), or
alternatively modify the NATSPublishUntilReceived function itself to accept and
use that count if intended; update all other call sites to match the chosen fix
so signatures remain consistent.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 54556b70-40b7-4be7-8d47-06c9c14adb50
📒 Files selected for processing (8)
router-tests/events/trigger_test.gorouter-tests/testenv/testenv.gorouter/pkg/pubsub/datasource/subscription_datasource.gorouter/pkg/pubsub/datasource/subscription_datasource_test.gorouter/pkg/pubsub/kafka/engine_datasource_factory.gorouter/pkg/pubsub/nats/adapter.gorouter/pkg/pubsub/nats/engine_datasource_factory.gorouter/pkg/pubsub/redis/engine_datasource_factory.go
…generation-on-cosmo-streams
endigma
requested changes
Jun 12, 2026
dkorittki
added a commit
to wundergraph/graphql-go-tools
that referenced
this pull request
Jun 12, 2026
Backwards compatible change, to let datasources implement the new interface `SubscriptionTriggerHasher`. If a datasource implements this interface the method `ProvideTriggerHashInput` is used by the subscription trigger generation to create a trigger id, instead of the `input`. `input` contains various data and it might not make sense for every data source to use it because it contains more data than needed for proper trigger deduplication. One such example is the pubsub datasource, implemented by the router. If a datasource does not implement this optional interface it falls back to the way it is before this change: Use `input`. Tests are provided in the router pull request, which uses the new interface. wundergraph/cosmo#2950 @coderabbitai summary ## Checklist - [x] I have discussed my proposed changes in an issue and have received approval to proceed. - [x] I have followed the coding standards of the project. - [ ] Tests or benchmarks have been added or updated. ## Open Source AI Manifesto This project follows the principles of the [Open Source AI Manifesto](https://human-oss.dev). Please ensure your contribution aligns with its principles. <!-- Please add any additional information or context regarding your changes here. --> --------- Co-authored-by: Jesse Thompson <34945114+endigma@users.noreply.github.com> Co-authored-by: linear-code[bot] <222613912+linear-code[bot]@users.noreply.github.com>
…generation-on-cosmo-streams
endigma
approved these changes
Jun 12, 2026
alepane21
approved these changes
Jun 15, 2026
…generation-on-cosmo-streams
dkorittki
deleted the
dominik/eng-9683-fix-inefficient-trigger-id-generation-on-cosmo-streams
branch
June 15, 2026 08:55
6 tasks
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
3 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.
Do not merge before wundergraph/graphql-go-tools#1531 is merged--> released in v2.4.6, go.mod is updatedThis pull request fixes a bug where the router creates more connections to message brokers (aka providers in Router terms) than necessary. The bug is only present on websocket subscriptions for Cosmo Streams. On GraphQL subscription subprotocol you can provide
initial_payload. The router uses this data alongside query extensions from the subscription GraphQL query to feed the engines trigger hash generator. It means different initial payloads or query extensions meant the router producing different triggers. A trigger causes the router to create a new connection to the message broker.For regular websocket subscriptions this makes sense because these two fields can influence the data a subgraph returns. You need to have different connections to have the router be able to receive different messages and deduplicate to the correct subscribers.
For Cosmo Streams this does not make sense, because the initial payload and query extensions don't influence a message brokers messages it sends to the router. Only the subject and provider-id do. It results in the router creating multiple triggers and thus connections to the broker, which all receive the same data. If clients subscribing to the router use individual headers and place them into
initial_payloadit means a 1 to 1 ratio of subscribers on the router and connections to Redis.To combat this there's a way now in the engine that allows datasources to provide the hash input by themselves. This pull request implements the interface for that on the pubsub datasource. Now we only use a provider-id and subject/topic/channel to generate the trigger id. This means every subscriber using a subscription which shares the same subject and provider-id shares the same trigger.
Summary by CodeRabbit
Checklist
Open Source AI Manifesto
This project follows the principles of the Open Source AI Manifesto. Please ensure your contribution aligns with its principles.