-
Notifications
You must be signed in to change notification settings - Fork 253
fix(router): fix inefficient trigger id generation #2950
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Merged
dkorittki
merged 15 commits into
main
from
dominik/eng-9683-fix-inefficient-trigger-id-generation-on-cosmo-streams
Jun 15, 2026
Merged
Changes from 9 commits
Commits
Show all changes
15 commits
Select commit
Hold shift + click to select a range
b11d365
fix: provide custom hasher to engine
dkorittki 17f15f7
chore: rename functions and methods
dkorittki 38554b3
chore: add type guard
dkorittki ebe268f
chore: add integration test to verify trigger deduplication
dkorittki 6d47194
chore: add test to verify trigger dedup on different initial payloads
dkorittki c9a2dce
Merge branch 'main' into dominik/eng-9683-fix-inefficient-trigger-id-…
dkorittki 1073c9a
chore: use temporary engine version (until merged)
dkorittki 5e3a3f0
chore: update to latest engine version
dkorittki 5d3a966
Merge branch 'main' into dominik/eng-9683-fix-inefficient-trigger-id-…
dkorittki 1851a3d
chore: align interface with the engines interface
dkorittki 3e845db
chore: rename interface method
dkorittki 3c5309b
chore: remove unnecessary parameter
dkorittki 59f4c25
chore: bump engine version
dkorittki 9dc95a4
Merge branch 'main' into dominik/eng-9683-fix-inefficient-trigger-id-…
dkorittki 98af68f
Merge branch 'main' into dominik/eng-9683-fix-inefficient-trigger-id-…
dkorittki File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,277 @@ | ||
| package events_test | ||
|
|
||
| import ( | ||
| "sync" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/stretchr/testify/require" | ||
| "github.com/wundergraph/cosmo/router-tests/testenv" | ||
| ) | ||
|
|
||
| // TestEDFSTriggerDeduplication verifies trigger ID generation for Cosmo Streams subscriptions. | ||
| // It tests that the trigger hash is based on the NATS subject and provider ID, not on | ||
| // the resolver input — meaning subscriptions to the same subject share one trigger | ||
| // even when they select different fields, have different headers or arguments. | ||
| func TestEDFSTriggerDeduplication(t *testing.T) { | ||
| t.Parallel() | ||
|
|
||
| // Two subscriptions with the same employeeID but different selected fields should share | ||
| // a single NATS trigger because both resolve to the same subject ("employeeUpdated.3"). | ||
| t.Run("same subject different selected fields shares one trigger", func(t *testing.T) { | ||
| t.Parallel() | ||
| testenv.Run(t, &testenv.Config{ | ||
| RouterConfigJSONTemplate: testenv.ConfigWithEdfsNatsJSONTemplate, | ||
| EnableNats: true, | ||
| }, func(t *testing.T, xEnv *testenv.Environment) { | ||
| var done sync.WaitGroup | ||
| done.Add(2) | ||
|
|
||
| go func() { | ||
| xEnv.WaitForSubscriptionCount(2, time.Second*10) | ||
| // Both subscriptions target the same NATS subject ("employeeUpdated.3") and | ||
| // provider ID, so they must resolve to the same trigger ID — wait for exactly | ||
| // one trigger to be initialized before asserting. | ||
| xEnv.WaitForTriggerCount(1, time.Second*10) | ||
| xEnv.RequireTriggerCount(1) | ||
| xEnv.NATSPublishUntilReceived(xEnv.NatsConnectionDefault, xEnv.GetPubSubName("employeeUpdated.3"), []byte(`{"id":3,"__typename":"Employee"}`), 2, time.Second*10) | ||
| }() | ||
|
|
||
| // Subscription 1: selects only id. | ||
| go func() { | ||
| defer done.Done() | ||
| conn := xEnv.InitGraphQLWebSocketConnection(nil, nil, nil) | ||
| defer conn.Close() | ||
|
|
||
| err := testenv.WSWriteJSON(t, conn, &testenv.WebSocketMessage{ | ||
| ID: "1", | ||
| Type: "subscribe", | ||
| Payload: []byte(`{"query":"subscription { employeeUpdated(employeeID: 3) { id } }"}`), | ||
| }) | ||
| require.NoError(t, err) | ||
|
|
||
| var msg testenv.WebSocketMessage | ||
| err = testenv.WSReadJSON(t, conn, &msg) | ||
| require.NoError(t, err) | ||
| require.Equal(t, "next", msg.Type) | ||
| require.Equal(t, "1", msg.ID) | ||
|
|
||
| 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) | ||
| require.NoError(t, err) | ||
| require.Equal(t, "complete", complete.Type) | ||
| require.Equal(t, "1", complete.ID) | ||
| }() | ||
|
|
||
| // Subscription 2: selects id and details — a different query shape over the same subject. | ||
| go func() { | ||
| defer done.Done() | ||
| conn := xEnv.InitGraphQLWebSocketConnection(nil, nil, nil) | ||
| defer conn.Close() | ||
|
|
||
| err := testenv.WSWriteJSON(t, conn, &testenv.WebSocketMessage{ | ||
| ID: "1", | ||
| Type: "subscribe", | ||
| Payload: []byte(`{"query":"subscription { employeeUpdated(employeeID: 3) { id details { forename surname } } }"}`), | ||
| }) | ||
| require.NoError(t, err) | ||
|
|
||
| var msg testenv.WebSocketMessage | ||
| err = testenv.WSReadJSON(t, conn, &msg) | ||
| require.NoError(t, err) | ||
| require.Equal(t, "next", msg.Type) | ||
| require.Equal(t, "1", msg.ID) | ||
|
|
||
| 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) | ||
| require.NoError(t, err) | ||
| require.Equal(t, "complete", complete.Type) | ||
| require.Equal(t, "1", complete.ID) | ||
| }() | ||
|
|
||
| done.Wait() | ||
| xEnv.WaitForSubscriptionCount(0, time.Second*5) | ||
| }) | ||
| }) | ||
|
|
||
| // Two subscriptions with the same query but different initial_payload (headers) should | ||
| // still share a single NATS trigger because the trigger ID is based on the NATS subject, | ||
| // not on connection-level metadata like headers. | ||
| t.Run("same subject different initial payload shares one trigger", func(t *testing.T) { | ||
| t.Parallel() | ||
| testenv.Run(t, &testenv.Config{ | ||
| RouterConfigJSONTemplate: testenv.ConfigWithEdfsNatsJSONTemplate, | ||
| EnableNats: true, | ||
| }, func(t *testing.T, xEnv *testenv.Environment) { | ||
| var done sync.WaitGroup | ||
| done.Add(2) | ||
|
|
||
| go func() { | ||
| xEnv.WaitForSubscriptionCount(2, time.Second*10) | ||
| xEnv.WaitForTriggerCount(1, time.Second*10) | ||
| xEnv.RequireTriggerCount(1) | ||
| xEnv.NATSPublishUntilReceived(xEnv.NatsConnectionDefault, xEnv.GetPubSubName("employeeUpdated.3"), []byte(`{"id":3,"__typename":"Employee"}`), 2, time.Second*10) | ||
| }() | ||
|
|
||
| // Subscription 1: sends Authorization header A in connection_init. | ||
| go func() { | ||
| defer done.Done() | ||
| conn := xEnv.InitGraphQLWebSocketConnection(nil, nil, []byte(`{"headers":{"Authorization":"Bearer token-a"}}`)) | ||
| defer conn.Close() | ||
|
|
||
| err := testenv.WSWriteJSON(t, conn, &testenv.WebSocketMessage{ | ||
| ID: "1", | ||
| Type: "subscribe", | ||
| Payload: []byte(`{"query":"subscription { employeeUpdated(employeeID: 3) { id } }"}`), | ||
| }) | ||
| require.NoError(t, err) | ||
|
|
||
| var msg testenv.WebSocketMessage | ||
| err = testenv.WSReadJSON(t, conn, &msg) | ||
| require.NoError(t, err) | ||
| require.Equal(t, "next", msg.Type) | ||
| require.Equal(t, "1", msg.ID) | ||
|
|
||
| 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) | ||
| require.NoError(t, err) | ||
| require.Equal(t, "complete", complete.Type) | ||
| require.Equal(t, "1", complete.ID) | ||
| }() | ||
|
|
||
| // Subscription 2: sends a different Authorization header in connection_init. | ||
| go func() { | ||
| defer done.Done() | ||
| conn := xEnv.InitGraphQLWebSocketConnection(nil, nil, []byte(`{"headers":{"Authorization":"Bearer token-b"}}`)) | ||
| defer conn.Close() | ||
|
|
||
| err := testenv.WSWriteJSON(t, conn, &testenv.WebSocketMessage{ | ||
| ID: "1", | ||
| Type: "subscribe", | ||
| Payload: []byte(`{"query":"subscription { employeeUpdated(employeeID: 3) { id } }"}`), | ||
| }) | ||
| require.NoError(t, err) | ||
|
|
||
| var msg testenv.WebSocketMessage | ||
| err = testenv.WSReadJSON(t, conn, &msg) | ||
| require.NoError(t, err) | ||
| require.Equal(t, "next", msg.Type) | ||
| require.Equal(t, "1", msg.ID) | ||
|
|
||
| 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) | ||
| require.NoError(t, err) | ||
| require.Equal(t, "complete", complete.Type) | ||
| require.Equal(t, "1", complete.ID) | ||
| }() | ||
|
|
||
| done.Wait() | ||
| xEnv.WaitForSubscriptionCount(0, time.Second*5) | ||
| }) | ||
| }) | ||
|
|
||
| // Two subscriptions with different employeeIDs resolve to different NATS subjects and | ||
| // must therefore receive independent triggers. | ||
| t.Run("different subjects use separate triggers", func(t *testing.T) { | ||
| t.Parallel() | ||
| testenv.Run(t, &testenv.Config{ | ||
| RouterConfigJSONTemplate: testenv.ConfigWithEdfsNatsJSONTemplate, | ||
| EnableNats: true, | ||
| }, func(t *testing.T, xEnv *testenv.Environment) { | ||
| var done sync.WaitGroup | ||
| done.Add(2) | ||
|
|
||
| go func() { | ||
| xEnv.WaitForSubscriptionCount(2, time.Second*10) | ||
| xEnv.WaitForTriggerCount(2, time.Second*10) | ||
| xEnv.NATSPublishUntilReceived(xEnv.NatsConnectionDefault, xEnv.GetPubSubName("employeeUpdated.3"), []byte(`{"id":3,"__typename":"Employee"}`), 1, time.Second*10) | ||
| xEnv.NATSPublishUntilReceived(xEnv.NatsConnectionDefault, xEnv.GetPubSubName("employeeUpdated.4"), []byte(`{"id":4,"__typename":"Employee"}`), 1, time.Second*10) | ||
| }() | ||
|
|
||
| // Subscription 1: employeeID 3 → subject "employeeUpdated.3" | ||
| go func() { | ||
| defer done.Done() | ||
| conn := xEnv.InitGraphQLWebSocketConnection(nil, nil, nil) | ||
| defer conn.Close() | ||
|
|
||
| err := testenv.WSWriteJSON(t, conn, &testenv.WebSocketMessage{ | ||
| ID: "1", | ||
| Type: "subscribe", | ||
| Payload: []byte(`{"query":"subscription { employeeUpdated(employeeID: 3) { id } }"}`), | ||
| }) | ||
| require.NoError(t, err) | ||
|
|
||
| var msg testenv.WebSocketMessage | ||
| err = testenv.WSReadJSON(t, conn, &msg) | ||
| require.NoError(t, err) | ||
| require.Equal(t, "next", msg.Type) | ||
| require.Equal(t, "1", msg.ID) | ||
|
|
||
| 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) | ||
| require.NoError(t, err) | ||
| require.Equal(t, "complete", complete.Type) | ||
| require.Equal(t, "1", complete.ID) | ||
| }() | ||
|
|
||
| // Subscription 2: employeeID 4 → subject "employeeUpdated.4" | ||
| go func() { | ||
| defer done.Done() | ||
| conn := xEnv.InitGraphQLWebSocketConnection(nil, nil, nil) | ||
| defer conn.Close() | ||
|
|
||
| err := testenv.WSWriteJSON(t, conn, &testenv.WebSocketMessage{ | ||
| ID: "1", | ||
| Type: "subscribe", | ||
| Payload: []byte(`{"query":"subscription { employeeUpdated(employeeID: 4) { id } }"}`), | ||
| }) | ||
| require.NoError(t, err) | ||
|
|
||
| var msg testenv.WebSocketMessage | ||
| err = testenv.WSReadJSON(t, conn, &msg) | ||
| require.NoError(t, err) | ||
| require.Equal(t, "next", msg.Type) | ||
| require.Equal(t, "1", msg.ID) | ||
|
|
||
| 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) | ||
| require.NoError(t, err) | ||
| require.Equal(t, "complete", complete.Type) | ||
| require.Equal(t, "1", complete.ID) | ||
| }() | ||
|
|
||
| done.Wait() | ||
| xEnv.WaitForSubscriptionCount(0, time.Second*5) | ||
| }) | ||
| }) | ||
| } |
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
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
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
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
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
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
Oops, something went wrong.
Oops, something went wrong.
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.
Uh oh!
There was an error while loading. Please reload this page.