fix(pubsub): fix inefficient trigger id generation

[dkorittki]

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

• I have discussed my proposed changes in an issue and have received approval to proceed.
• 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. Please ensure your contribution aligns with its principles.

[coderabbitai]

Note

Reviews paused

It 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 reviews.auto_review.auto_pause_after_reviewed_commits setting.

Use the following commands to manage reviews:

• @coderabbitai resume to resume automatic reviews.
• @coderabbitai review to trigger a single review.

Use the checkboxes below for quick actions:

• ▶️ Resume reviews
• 🔍 Trigger review

📝 Walkthrough

Walkthrough

Adds xxhash import and a new SubscriptionDataSource.HashTriggerInput method; Resolver.prepareTrigger now accepts the data source and uses its HashTriggerInput to seed the trigger hash before finalizing triggerID; both sync and async subscription resolvers pass the source into prepareTrigger.

Changes

Subscription Trigger Hashing

Layer / File(s)	Summary
SubscriptionDataSource HashTriggerInput contract v2/pkg/engine/resolve/datasource.go	Adds github.com/cespare/xxhash/v2 import and a HashTriggerInput(ctx *Context, input []byte, xxh *xxhash.Digest) error method on SubscriptionDataSource for datasources to write identity-relevant bytes into an xxhash.Digest.
Resolver trigger-hash generation with source hook v2/pkg/engine/resolve/resolve.go	Resolver.prepareTrigger now accepts the subscription SubscriptionDataSource, calls source.HashTriggerInput(ctx, input, keyGen) to seed the xxhash digest, optionally appends subgraph headers hash, computes triggerID from the digest, and returns the generator to the pool; both sync and async subscription resolvers pass subscription.Trigger.Source into prepareTrigger and handle prepare errors by writing a failed subscription response.
Datasource implementation and test doubles v2/pkg/engine/datasource/graphql_datasource/graphql_datasource.go, v2/pkg/engine/resolve/resolve_test.go, v2/pkg/engine/resolve/resolver_subscription_test.go	Implements HashTriggerInput on SubscriptionSource (writes input into provided *xxhash.Digest) and adds equivalent HashTriggerInput implementations to in-test fake streams and FakeSource so test doubles conform to the new interface.

Estimated code review effort

🎯 3 (Moderate) | ⏱️ ~20 minutes

🚥 Pre-merge checks | ✅ 4

✅ Passed checks (4 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title clearly and specifically describes the main change: fixing inefficient trigger ID generation in pub/sub subscriptions.
Description check	✅ Passed	The description is directly related to the changeset, explaining the backwards-compatible interface addition and how datasources can implement it for more efficient trigger ID generation.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch dominik/eng-9714-fix-inefficient-trigger-id-generation-in-engine

---

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[dkorittki]

@coderabbitai review

[coderabbitai]

✅ Action performed

Review finished.

Note: CodeRabbit is an incremental review system and does not re-review already reviewed commits. This command is applicable only when automatic reviews are paused.

[coderabbitai]

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 `@v2/pkg/engine/resolve/resolve.go`:
- Around line 1283-1295: The custom hasher path
(SubscriptionTriggerHasher.ProvideTriggerHashInput) may have written partial
bytes into keyGen before returning an error, so the fallback that writes raw
input into the same keyGen will mix partial custom data with raw input; to fix
it, ensure keyGen is reset to a clean state before performing the raw-input
fallback: detect when ProvideTriggerHashInput returns a non-nil error and either
call keyGen.Reset()/Reinitialize the hasher or create a fresh hasher instance
before executing keyGen.Write(input), keeping the existing usedHasher boolean
logic intact (refer to ProvideTriggerHashInput, SubscriptionTriggerHasher and
keyGen).

🪄 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: Path: .coderabbit.yaml

Review profile: CHILL

Plan: Pro

Run ID: f6fdbf3e-8197-40db-ae84-25f14fecf445

📥 Commits

Reviewing files that changed from the base of the PR and between 55609f7 and aea8a60.

📒 Files selected for processing (2)

• v2/pkg/engine/resolve/datasource.go
• v2/pkg/engine/resolve/resolve.go

[coderabbitai]

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 `@v2/pkg/engine/resolve/resolve.go`:
- Line 1281: The call to source.TriggerIDInput currently discards its returned
error; update prepareTrigger to capture and propagate that error (e.g., err :=
source.TriggerIDInput(ctx, input, keyGen); if err != nil { return nil, err } or
return err consistent with prepareTrigger's signature) and then update its
callers (the functions that invoke prepareTrigger) to handle the propagated
error (either return it up the stack or log with context and continue). If you
prefer not to change the signature, at minimum log the error with context
(including trigger/identity info) instead of using a blank assignment; reference
source.TriggerIDInput and prepareTrigger to locate the call and adjust the two
call sites that currently assume no error.

🪄 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: Path: .coderabbit.yaml

Review profile: CHILL

Plan: Pro

Run ID: e916087f-38d7-42a1-a5f4-32ade3354538

📥 Commits

Reviewing files that changed from the base of the PR and between 46b36e8 and e9e1c00.

📒 Files selected for processing (5)

• v2/pkg/engine/datasource/graphql_datasource/graphql_datasource.go
• v2/pkg/engine/resolve/datasource.go
• v2/pkg/engine/resolve/resolve.go
• v2/pkg/engine/resolve/resolve_test.go
• v2/pkg/engine/resolve/resolver_subscription_test.go

[coderabbitai]

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 `@v2/pkg/engine/resolve/datasource.go`:
- Around line 21-23: prepareTrigger currently ignores the error returned by
HashTriggerInput (call site: _ = source.HashTriggerInput(...)), violating the
contract; change prepareTrigger to capture the error from
source.HashTriggerInput(ctx, input, xxh), return that error (or wrap it with
context) and abort subscription creation so triggerID is not computed when
HashTriggerInput fails; update prepareTrigger's signature and its callers to
propagate the error up the call chain (or, alternatively, if failures are
impossible, remove the error return from HashTriggerInput and update its
implementations accordingly), referencing the HashTriggerInput method and
prepareTrigger function and ensuring any code computing triggerID only runs when
no error is returned.

🪄 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: Path: .coderabbit.yaml

Review profile: CHILL

Plan: Pro

Run ID: 33546748-5a1b-486f-aee2-26023a45b9b3

📥 Commits

Reviewing files that changed from the base of the PR and between e9e1c00 and a83cd53.

📒 Files selected for processing (5)

• v2/pkg/engine/datasource/graphql_datasource/graphql_datasource.go
• v2/pkg/engine/resolve/datasource.go
• v2/pkg/engine/resolve/resolve.go
• v2/pkg/engine/resolve/resolve_test.go
• v2/pkg/engine/resolve/resolver_subscription_test.go

🚧 Files skipped from review as they are similar to previous changes (2)

• v2/pkg/engine/resolve/resolve_test.go
• v2/pkg/engine/resolve/resolve.go