feat(engine): idempotency#4045
Open
mrkaye97 wants to merge 76 commits into
Open
Conversation
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
github-actions
Bot
added
sdk-py
mrkaye97
commented
Jun 4, 2026
Comment on lines
+38
to
+72
| function extractExistingRunIdFromGrpcError(e: ServiceError): string { | ||
| try { | ||
| const [binData] = e.metadata.get('grpc-status-details-bin'); | ||
| if (!binData) return ''; | ||
|
|
||
| const status = RpcStatus.decode(binData instanceof Buffer ? binData : Buffer.from(binData)); | ||
| for (const detail of status.details) { | ||
| if (detail.typeUrl.includes('IdempotencyCollisionError')) { | ||
| return IdempotencyCollisionErrorProto.decode(detail.value).existingRunExternalId ?? ''; | ||
| } | ||
| } | ||
| } catch { | ||
| // ignore decoding errors | ||
| } | ||
| return ''; | ||
| } | ||
|
|
||
| function isNiceGrpcAlreadyExists(e: unknown): e is ClientError { | ||
| return e instanceof ClientError && e.code === NiceGrpcStatus.ALREADY_EXISTS; | ||
| } | ||
|
|
||
| function extractRunIdFromNiceGrpcMetadata(metadata: NiceGrpcMetadata | undefined): string { | ||
| if (!metadata) return ''; | ||
| try { | ||
| const binData = metadata.get('grpc-status-details-bin'); | ||
| if (!binData) return ''; | ||
| const status = RpcStatus.decode(binData); | ||
| for (const detail of status.details) { | ||
| if (detail.typeUrl.includes('IdempotencyCollisionError')) { | ||
| return IdempotencyCollisionErrorProto.decode(detail.value).existingRunExternalId ?? ''; | ||
| } | ||
| } | ||
| } catch { | ||
| // ignore decoding errors | ||
| } |
Contributor
Author
There was a problem hiding this comment.
@grutt do you have any idea how to do this in a less insane way? there's a library in python, but for Ruby + TS I needed to vendor this RpcStatus proto and use it here
Contributor
There was a problem hiding this comment.
Pull request overview
Adds first-class, user-configurable idempotency keys for workflows/standalone tasks end-to-end (DB + engine trigger paths + SDKs), so duplicate triggers can be rejected (SDK error) or swallowed (event path), preventing duplicate runs within a TTL window.
Changes:
- Adds workflow version–level idempotency config (
expression,ttl_ms) to the API contracts, DB schema/migrations, and engine repository/scheduler trigger flows. - Implements SDK support (TypeScript/Ruby/Python/Go) including surfaced collision errors (e.g.
IdempotencyCollisionError) and updated examples/e2e tests/docs. - Updates idempotency-key claiming logic and related scheduling/trigger plumbing, plus a few adjacent OLAP/API model improvements needed by the new behavior.
Reviewed changes
Copilot reviewed 125 out of 129 changed files in this pull request and generated 6 comments.
Show a summary per file
| File | Description |
|---|---|
| sql/schema/v1-olap.sql | Extend CEL failure source enum for idempotency failures |
| sql/schema/v0.sql | Persist idempotency config on WorkflowVersion |
| sdks/typescript/src/v1/task.ts | Add TS idempotency config type |
| sdks/typescript/src/v1/index.ts | Export idempotency collision error |
| sdks/typescript/src/v1/examples/idempotency/workflow.ts | TS SDK idempotency example workflow |
| sdks/typescript/src/v1/examples/idempotency/run.ts | TS SDK idempotency collision handling example |
| sdks/typescript/src/v1/examples/idempotency/idempotency.e2e.ts | TS e2e coverage for idempotency behavior |
| sdks/typescript/src/v1/examples/e2e-worker.ts | Register TS idempotency workflows in worker |
| sdks/typescript/src/v1/declaration.ts | Add idempotency option to workflow/task declarations |
| sdks/typescript/src/v1/client/worker/worker-internal.ts | Send idempotency config on workflow registration |
| sdks/typescript/src/v1/client/admin.ts | Map ALREADY_EXISTS to IdempotencyCollisionError |
| sdks/typescript/src/util/retrier.ts | Add shouldRetry hook to avoid retrying collisions |
| sdks/typescript/src/util/errors/idempotency-collision-error.ts | New TS error type for idempotency collisions |
| sdks/typescript/src/protoc/v1/workflows.ts | Generated proto changes for idempotency fields/errors |
| sdks/typescript/src/protoc/google/rpc/status.ts | Vendor generated google.rpc.Status for details decoding |
| sdks/typescript/src/protoc/google/protobuf/any.ts | Vendor generated google.protobuf.Any dependency |
| sdks/typescript/package.json | Bump TS SDK version |
| sdks/typescript/generate-protoc.sh | Include vendored google protos in generation |
| sdks/typescript/CHANGELOG.md | Document idempotency feature in TS SDK |
| sdks/ruby/src/lib/hatchet/workflow.rb | Add Ruby workflow idempotency config + proto mapping |
| sdks/ruby/src/lib/hatchet/version.rb | Bump Ruby SDK version |
| sdks/ruby/src/lib/hatchet/exceptions.rb | Add Ruby IdempotencyCollisionError |
| sdks/ruby/src/lib/hatchet/clients/grpc/admin.rb | Parse grpc status details to raise collision error |
| sdks/ruby/src/lib/hatchet-sdk.rb | Wire task wrapper to pass idempotency through |
| sdks/ruby/src/hatchet-sdk.gemspec | Add dependency for google RPC status decoding |
| sdks/ruby/src/Gemfile.lock | Lockfile updates for new Ruby dependency/version |
| sdks/ruby/src/CHANGELOG.md | Document idempotency feature in Ruby SDK |
| sdks/ruby/src/.rubocop.yml | Rubocop config adjustment |
| sdks/ruby/examples/idempotency/worker.rb | Ruby idempotency example worker |
| sdks/ruby/examples/idempotency/trigger.rb | Ruby idempotency example trigger |
| sdks/ruby/examples/idempotency/test_idempotency_spec.rb | Ruby example-based idempotency tests |
| sdks/ruby/examples/Gemfile.lock | Examples lockfile updates |
| sdks/python/pyproject.toml | Add grpc-status deps + bump Python SDK version |
| sdks/python/hatchet_sdk/types/idempotency.py | New Python idempotency config type |
| sdks/python/hatchet_sdk/runnables/workflow.py | Attach idempotency to workflow registration; input serialization tweak |
| sdks/python/hatchet_sdk/runnables/types.py | Plumb idempotency into workflow config model |
| sdks/python/hatchet_sdk/hatchet.py | Add idempotency args to workflow/task APIs |
| sdks/python/hatchet_sdk/exceptions.py | Add Python IdempotencyCollisionError |
| sdks/python/hatchet_sdk/contracts/v1/workflows_pb2.pyi | Generated typing updates for idempotency messages |
| sdks/python/hatchet_sdk/config.py | Add configurable NOT_FOUND retry toggle |
| sdks/python/hatchet_sdk/clients/rest/tenacity_utils.py | Retry policy updates (incl. NOT_FOUND configurability) |
| sdks/python/hatchet_sdk/clients/rest/models/worker.py | Tighten worker status typing to enum |
| sdks/python/hatchet_sdk/clients/rest/models/worker_status.py | New enum model for worker status |
| sdks/python/hatchet_sdk/clients/rest/models/v1_wait_item.py | New durable wait item model |
| sdks/python/hatchet_sdk/clients/rest/models/v1_durable_wait_condition.py | New durable wait condition model |
| sdks/python/hatchet_sdk/clients/rest/models/v1_durable_wait_condition_kind.py | Add CHILD_WORKFLOW kind |
| sdks/python/hatchet_sdk/clients/rest/models/v1_durable_event_log_entry.py | Model durable wait payload changes |
| sdks/python/hatchet_sdk/clients/rest/models/tenant.py | Add server_url/region fields |
| sdks/python/hatchet_sdk/clients/rest/models/feature_flag_id.py | Add new feature flag IDs |
| sdks/python/hatchet_sdk/clients/rest/models/init.py | Export newly generated REST models |
| sdks/python/hatchet_sdk/clients/rest/api/worker_api.py | Add worker list filters (offset/limit/statuses) |
| sdks/python/hatchet_sdk/clients/rest/api/webhook_api.py | Accept 204 response for webhook receive |
| sdks/python/hatchet_sdk/clients/rest/api/tenant_api.py | Add taskNames query param support |
| sdks/python/hatchet_sdk/clients/rest/api/rate_limits_api.py | Add rate limit delete endpoint client |
| sdks/python/hatchet_sdk/clients/rest/init.py | Export newly generated REST models |
| sdks/python/hatchet_sdk/clients/admin.py | Map ALREADY_EXISTS + status details to collision error; refactor client getters |
| sdks/python/hatchet_sdk/init.py | Export idempotency types/errors |
| sdks/python/examples/worker.py | Register Python idempotency workflows in example worker |
| sdks/python/examples/idempotency/worker.py | Python idempotency example worker |
| sdks/python/examples/idempotency/trigger.py | Python idempotency example trigger |
| sdks/python/examples/idempotency/test_idempotency.py | Python idempotency tests |
| sdks/python/examples/events/test_event.py | Narrow event test listing via workflow_ids |
| sdks/python/CHANGELOG.md | Document idempotency feature in Python SDK |
| sdks/go/workflow.go | Add Go SDK idempotency config + collision mapping |
| sdks/go/internal/declaration.go | Include idempotency in workflow dump/register request |
| sdks/go/examples/idempotency/worker.go | Go SDK idempotency example worker (package) |
| sdks/go/examples/idempotency/trigger.go | Go SDK idempotency example trigger (package) |
| sdks/go/errors.go | Add Go IdempotencyCollisionError |
| sdks/go/e2e/idempotency_test.go | Go e2e tests for idempotency |
| pkg/scheduling/v1/tenant_manager.go | Return idempotency collisions from optimistic scheduling |
| pkg/scheduling/v1/pool.go | Plumb collisions through scheduling pool API |
| pkg/repository/workflow.go | Persist idempotency config when creating workflow versions |
| pkg/repository/sqlcv1/workflows.sql.go | sqlc generated: store/read idempotency columns |
| pkg/repository/sqlcv1/workflows.sql | sqlc query: insert idempotency fields |
| pkg/repository/sqlcv1/triggers.sql.go | sqlc generated: include idempotency fields in trigger lookups |
| pkg/repository/sqlcv1/triggers.sql | sqlc query: include idempotency fields in trigger lookups |
| pkg/repository/sqlcv1/tasks.sql.go | sqlc generated: normalize zero UUID filter_id to NULL |
| pkg/repository/sqlcv1/tasks.sql | sqlc query: normalize zero UUID filter_id to NULL |
| pkg/repository/sqlcv1/olap.sql.go | sqlc generated: adjust event lookup filtering query |
| pkg/repository/sqlcv1/olap.sql | sqlc query: adjust event lookup filtering query |
| pkg/repository/sqlcv1/models.go | sqlc models: add enum value + workflow version idempotency fields |
| pkg/repository/sqlcv1/idempotency-keys.sql.go | sqlc generated: updated key-claiming query + params |
| pkg/repository/sqlcv1/idempotency-keys.sql | Rewrite claim logic for idempotency keys |
| pkg/repository/scheduler.go | Optimistic repo interface updated to return collisions |
| pkg/repository/scheduler_optimistic.go | Propagate collision list from trigger paths |
| pkg/repository/olap.go | Make event scope nullable; adjust event data population window |
| pkg/repository/ids.go | Replace explicit idempotency key field with presence flag |
| pkg/repository/idempotency.go | Replace create-only API with ClaimKey + collision struct |
| pkg/repository/durable_events.go | Track/signal CEL evaluation failures on durable triggers |
| pkg/client/create/tasks.go | Legacy Go client: add idempotency option in create opts |
| pkg/client/admin.go | Legacy Go client: map grpc status detail to idempotency violation |
| internal/services/ticker/schedule_workflow_v1.go | Use new idempotency claim API for scheduled workflows |
| internal/services/scheduler/v1/optimistic.go | Return idempotency collisions from optimistic scheduling service |
| internal/services/dispatcher/v1/server.go | Signal CEL evaluation failures on durable trigger ingestion |
| internal/services/controllers/task/trigger/trigger.go | Return collisions + signal CEL failures during trigger |
| internal/services/controllers/task/controller.go | Adapt to new TriggerFromWorkflowNames return signature |
| internal/services/controllers/olap/controller.go | Preserve nullable filterId when creating event triggers |
| internal/services/admin/v1/server.go | Return grpc ALREADY_EXISTS with collision details |
| internal/services/admin/v1/admin.go | Always construct TriggerWriter; gate grpc usage by flag |
| internal/services/admin/server_v1.go | Return grpc ALREADY_EXISTS with collision details (v1 server) |
| internal/services/admin/admin.go | Always construct TriggerWriter; gate grpc usage by flag |
| hack/proto/vendor/google/rpc/status.proto | Vendor google rpc status proto |
| frontend/docs/pages/v1/idempotency.mdx | New docs page for idempotency feature |
| frontend/docs/pages/v1/concurrency.mdx | Minor docs punctuation fix |
| frontend/docs/pages/v1/_meta.js | Add idempotency page to docs nav |
| frontend/docs/pages/reference/changelog/typescript.mdx | Sync TS changelog into docs |
| frontend/docs/pages/reference/changelog/ruby.mdx | Sync Ruby changelog into docs |
| frontend/docs/pages/reference/changelog/python.mdx | Sync Python changelog into docs |
| examples/typescript/idempotency/workflow.ts | Repo-level TS idempotency example |
| examples/typescript/idempotency/run.ts | Repo-level TS collision handling example |
| examples/typescript/e2e-worker.ts | Repo-level TS example worker registration |
| examples/ruby/idempotency/worker.rb | Repo-level Ruby idempotency example |
| examples/ruby/idempotency/trigger.rb | Repo-level Ruby collision handling example |
| examples/python/worker.py | Repo-level Python example worker registration |
| examples/python/idempotency/worker.py | Repo-level Python idempotency example |
| examples/python/idempotency/trigger.py | Repo-level Python collision handling example |
| examples/go/idempotency/worker/main.go | Repo-level Go idempotency example (worker) |
| examples/go/idempotency/worker.go | Repo-level Go idempotency example (helper) |
| examples/go/idempotency/trigger/main.go | Repo-level Go idempotency example (trigger) |
| examples/go/idempotency/trigger.go | Repo-level Go idempotency example (helper) |
| cmd/hatchet-migrate/migrate/migrations/20260529180517_v1_0_116.sql | Migration adding workflow idempotency columns + enum value |
| api/v1/server/oas/transformers/v1/events.go | Fix event scope nullability in OAS transformer |
| api-contracts/v1/workflows.proto | Add idempotency config + collision error messages |
|
Promptless prepared a documentation update related to this change. Triggered by PR #4045 Created comprehensive documentation for the new idempotency feature including a conceptual guide explaining when to use idempotency, how it works, TTL configuration guidance, collision handling with SDK examples for all four languages (Python, TypeScript, Go, Ruby), and best practices for building effective idempotency expressions. Review: Document idempotency feature |
Contributor
|
|
mrkaye97
commented
Jul 2, 2026
Comment on lines
+160
to
+161
| pubBuffer = msgqueue.NewMQPubBuffer(opts.mq) | ||
| tw := trigger.NewTriggerWriter(opts.mq, opts.repo, opts.l, pubBuffer, slots, opts.promGate) |
Contributor
Author
There was a problem hiding this comment.
todo: check if these changes are correct - was a weird merge (looks like we also have duped pub buffers now
Contributor
|
|
Contributor
|
|
Contributor
|
|
Contributor
|
|
Contributor
|
|
Contributor
|
|
Contributor
|
|
Contributor
|
|
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
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
Adds support for user-facing idempotency configuration on tasks and workflows, to enable e.g. duplicate event pushes to not trigger duplicate tasks. Idempotency is configured on the task / workflow, and then collisions raise an error (which can be handled) on the SDKs when they occur on the trigger path. On the event path, the runs that would've been triggered are dropped.
Fixes #3673
Type of change
Checklist
Changes have been:
🤖 AI Disclosure