test(dispatcher): guard tenant-stream msg IDs against drift - #4547
Merged
Conversation
The pub/sub split (#4480) made the implicit durable-send mirror to tenant streams an explicit allowlist (msgqueue.tenantStreamMsgIDs), which nearly dropped task-cancelled on the API cancellation path during review. Nothing kept that allowlist in sync with what the dispatcher's streams consume. Make the dispatcher the checkable source of truth: msgsToWorkflowEvent's switch becomes a lookup in a package-level workflowEventConverters map and isMatchingWorkflowRunV1's switch a lookup in workflowRunMatchers, preserving behavior. msgqueue exports the allowlist via TenantStreamMsgIDs(), and TestTenantStreamMsgIDsInSync asserts set equality between the two, with failure messages saying exactly what to update on either side.
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
Contributor
Benchmark resultsCompared against |
…wn file Move workflowEventConverters, workflowRunMatchers, msgsToWorkflowEvent and isMatchingWorkflowRunV1 out of server.go into tenant_stream_events.go, next to the drift test that asserts their keys against msgqueue's allowlist. Collapse the repeated decode/loop/append scaffolding into two small generic adapters (eventConverter, runMatcher), so each map entry is just its payload type and field mapping; the workflow-run-finished status switch stays inside its closure. Drop the unused DispatcherImpl receivers from both functions. No behavior change: map misses and empty match results return exactly what they did before.
igor-kupczynski
marked this pull request as ready for review
July 28, 2026 10:10
abelanger5
approved these changes
Jul 28, 2026
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.
Description
Follow-up to #4480, which turned the implicit durable-send mirror to tenant streams into an explicit publish allowlist (
msgqueue.tenantStreamMsgIDs). During that review thetask-cancelledAPI cancellation path nearly fell out of the allowlist silently — nothing kept it in sync with what the dispatcher's streams actually consume, in either direction (a consumed ID missing from the allowlist is never published; an allowlisted ID nobody consumes is wasted publishes).TestTenantStreamMsgIDsInSyncasserts set equality betweenmsgqueue.TenantStreamMsgIDs()(new accessor) and the message IDs the dispatcher's streams consume, with failure messages saying exactly what to update on which side.workflowEventConvertersandworkflowRunMatchers— extracted together withmsgsToWorkflowEvent/isMatchingWorkflowRunV1(now receiver-less free functions) intotenant_stream_events.go, next to the test. The repeated decode/loop/append scaffolding is collapsed into two small generic adapters (eventConverter,runMatcher), so each map entry is just its payload type and field mapping.server.go−206).flowchart LR P[producers<br/>PubTenantMessage] -->|gated by<br/>tenantStreamMsgIDs| S[tenant stream] S --> C[workflowEventConverters] S --> M[workflowRunMatchers] T[TestTenantStreamMsgIDsInSync] -. asserts allowlist == consumer keys .- C T -.- MType of change
Checklist
Changes have been:
go test ./internal/msgqueue/ ./internal/services/dispatcher/passes; the drift test verified to fail with an actionable message when an ID is removed from either side🤖 AI Disclosure
I acknowledge that an LLM was used in the creation of this Pull Request, in accordance with Hatchet's AI_POLICY.md.
Details: Claude w/ Fable