Skip to content

Commit 34c8b92

Browse files
mrkaye97mrkaye97
authored
feat(engine): idempotency (#4045)
* fix: small optimization, no need to query if there are no claims * feat: add idempotency key expr to proto, workflow version * feat: wiring * fix: wiring to db * feat: add ttl col * chore: gen a whole bunch of python * feat: wire up idempotency config on the sdk * fix: engine wiring * fix: py type * chore: revert worker changes * feat: new example * feat: fix migration, add cols * feat: wiring * feat: wire up idempotency to the triggerTuple * feat: initial engine wiring * fix: len * fix: types * feat: add test * chore: rm unused deps * fix: use the v1 trigger endpoint on python * fix: make the python test do a real thing * feat: first pass bypassing mq fallback for idempotency * fix: remove todo, return the same id * feat: wire up existing run ids * fix: use test run id to dedupe across test runs * fix: improve test a bit * chore: lock * fix: add grpcio types * fix: drive by fix for a flaky test caused by event condition race * fix: not found behavior on the get_details method * fix: revert python changes * fix: handle collisions in the v1 ingest endpoint 🤦 * chore: gen * chore: add back some comments to shrink diff * chore: python version, changelog * fix: use the correct tx * fix: couple query bugs * fix: couple more event changes * feat: idempotency test with events * fix: type issue * fix: tests, lint * fix: test cleanup * fix: handle deduplication in the query * fix: test, gen * fix: conflict * fix: add prefix to idempotency key claim, clean up a tiny bit * fix: migration name, add idempotency key to cel eval failure source enum * fix: pr comments * chore: gen * feat: snippets, initial doc work * chore: vibe code other sdks * chore: vendor the status proto for js * fix: go docs * fix: ruby simplification * chore: changelogs * fix: ruby ci * fix: more test fixes * chore: gen * chore: gen again * fix: tests * chore: gen * chore: changelogs * chore: lockfile * chore: lint * chore: appease the cop * chore: fix copilot comments * chore: gen * chore: migration version * chore: migration version * fix: ordering to prevent deadlocks * fix: add validation, add skeleton for handling collisions on bulk trigger * feat: add key to v1_task * feat: wiring * feat: add idempotency keys to runs, dags, etc. tables * feat: wire up sending idempotency back over the api to the dash * feat: wire up idempotency filter and column on the fe * feat: return partial success on bulk trigger via an error * feat: wrap bulk triggers * feat: add test * feat: go migration for indexes * fix: order * fix: down * fix: weird diff * fix: always create trigger writer * chore: gen other sdks * fix: queries * fix: param ordering * chore: gen * chore: prettier * fix: merge * fix: migration * chore: lint * chore: bundle * chore: gen changelogs * feat: add TTL-based idempotency config + ABC to allow for better extensibility for other strategies * chore: other sdks, gen * chore: lint * fix: rubocop * fix: rm generic * chore: schema
1 parent 30a9afa commit 34c8b92

143 files changed

Lines changed: 5736 additions & 1196 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎api-contracts/openapi/components/schemas/v1/task.yaml‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -121,6 +121,9 @@ V1TaskSummary:
121121
type: string
122122
description: The external ID of the parent task.
123123
format: uuid
124+
idempotencyKey:
125+
type: string
126+
description: The idempotency key that was claimed by the task run
124127
required:
125128
- metadata
126129
- createdAt

‎api-contracts/openapi/paths/v1/workflow-runs/workflow_run.yaml‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -119,6 +119,14 @@ listWorkflowRuns:
119119
required: false
120120
schema:
121121
$ref: "../../../components/schemas/_index.yaml#/V1RunningFilter"
122+
- description: The idempotency key(s) to filter for
123+
in: query
124+
name: idempotency_keys
125+
required: false
126+
schema:
127+
type: array
128+
items:
129+
type: string
122130
responses:
123131
"200":
124132
content:

‎api-contracts/v1/workflows.proto‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -111,8 +111,24 @@ message CreateWorkflowVersionRequest {
111111
repeated Concurrency concurrency_arr = 12; // (optional) the workflow concurrency options
112112
repeated DefaultFilter default_filters = 13; // (optional) the default filters for the workflow
113113
optional bytes input_json_schema = 14; // (optional) the JSON schema for the workflow input
114+
115+
optional IdempotencyConfig idempotency = 15; // (optional) idempotency configuration for the workflow
116+
}
117+
118+
message IdempotencyConfig {
119+
string expression = 1; // a CEL expression for determining the idempotency key for workflow runs
120+
int64 ttl_ms = 2; // time-to-live for idempotency keys in milliseconds
114121
}
115122

123+
message IdempotencyCollisionError {
124+
string existing_run_external_id = 1; // the external ID of the existing workflow run that caused the collision
125+
string colliding_run_external_id = 2; // the external ID of the workflow run that caused the collision
126+
}
127+
128+
message BulkTriggerIdempotencyCollisionError {
129+
repeated string successful_workflow_run_external_ids = 1; // the external IDs of the successfully triggered workflow runs
130+
repeated IdempotencyCollisionError collisions = 2; // the idempotency collision errors
131+
}
116132

117133
message DefaultFilter {
118134
string expression = 1; // (required) the CEL expression for the filter

‎api/v1/server/handlers/v1/workflow-runs/list.go‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -140,6 +140,7 @@ func (t *V1WorkflowRunsService) WithDags(ctx context.Context, request gen.V1Work
140140
Offset: offset,
141141
IncludePayloads: includePayloads,
142142
UseGinIndex: useGinIndex,
143+
IdempotencyKeys: request.Params.IdempotencyKeys,
143144
}
144145

145146
additionalMetadataFilters := make(map[string]interface{})
@@ -291,6 +292,7 @@ func (t *V1WorkflowRunsService) OnlyTasks(ctx context.Context, request gen.V1Wor
291292
WorkerId: request.Params.WorkerId,
292293
IncludePayloads: includePayloads,
293294
UseGinIndex: useGinIndex,
295+
IdempotencyKeys: request.Params.IdempotencyKeys,
294296
}
295297

296298
additionalMetadataFilters := make(map[string]interface{})

‎api/v1/server/oas/gen/openapi.gen.go‎

Lines changed: 237 additions & 223 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎api/v1/server/oas/transformers/v1/events.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -77,7 +77,7 @@ func ToV1Event(event *v1.EventWithPayload) gen.V1Event {
7777
},
7878
Payload: &payload,
7979
SeenAt: &event.EventSeenAt.Time,
80-
Scope: &event.EventScope,
80+
Scope: event.EventScope,
8181
TriggeredRuns: &triggeredRuns,
8282
TriggeringWebhookName: event.TriggeringWebhookName,
8383
}

‎api/v1/server/oas/transformers/v1/workflow_runs.go‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -105,6 +105,7 @@ func WorkflowRunDataToV1TaskSummary(task *v1.WorkflowRunData, workflowIdsToNames
105105
RetryCount: &retryCount,
106106
Attempt: &attempt,
107107
ParentTaskExternalId: parentTaskExternalId,
108+
IdempotencyKey: task.IdempotencyKey,
108109
}
109110

110111
if isEvicted {
@@ -192,6 +193,12 @@ func PopulateTaskRunDataRowToV1TaskSummary(task *v1.TaskWithPayloads, workflowNa
192193

193194
taskStatus, isEvicted := mapOlapStatus(string(task.Status))
194195

196+
var idempotencyKey *string
197+
198+
if task.IdempotencyKey.Valid {
199+
idempotencyKey = &task.IdempotencyKey.String
200+
}
201+
195202
summary := gen.V1TaskSummary{
196203
Metadata: gen.APIResourceMeta{
197204
Id: task.ExternalID.String(),
@@ -223,6 +230,7 @@ func PopulateTaskRunDataRowToV1TaskSummary(task *v1.TaskWithPayloads, workflowNa
223230
Attempt: &attempt,
224231
WorkflowRunExternalId: task.WorkflowRunID,
225232
ParentTaskExternalId: task.ParentTaskExternalID,
233+
IdempotencyKey: idempotencyKey,
226234
}
227235

228236
if isEvicted {

‎cmd/hatchet-migrate/migrate/migrations/20260715180517_v1_0_129.sql‎

Lines changed: 265 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,265 @@
1+
-- +goose Up
2+
-- +goose StatementBegin
3+
ALTER TABLE "WorkflowVersion"
4+
ADD COLUMN "idempotencyKeyExpression" TEXT,
5+
ADD COLUMN "idempotencyKeyTtlMs" BIGINT
6+
;
7+
8+
ALTER TYPE v1_cel_evaluation_failure_source ADD VALUE IF NOT EXISTS 'IDEMPOTENCY_KEY';
9+
10+
ALTER TABLE v1_task ADD COLUMN idempotency_key TEXT;
11+
ALTER TABLE v1_dag ADD COLUMN idempotency_key TEXT;
12+
13+
ALTER TABLE v1_tasks_olap ADD COLUMN idempotency_key TEXT;
14+
ALTER TABLE v1_dags_olap ADD COLUMN idempotency_key TEXT;
15+
ALTER TABLE v1_runs_olap ADD COLUMN idempotency_key TEXT;
16+
17+
CREATE OR REPLACE FUNCTION v1_tasks_olap_insert_function()
18+
RETURNS TRIGGER AS
19+
$$
20+
BEGIN
21+
INSERT INTO v1_runs_olap (
22+
tenant_id,
23+
id,
24+
inserted_at,
25+
external_id,
26+
readable_status,
27+
kind,
28+
workflow_id,
29+
workflow_version_id,
30+
additional_metadata,
31+
parent_task_external_id,
32+
idempotency_key
33+
)
34+
SELECT
35+
tenant_id,
36+
id,
37+
inserted_at,
38+
external_id,
39+
readable_status,
40+
'TASK',
41+
workflow_id,
42+
workflow_version_id,
43+
additional_metadata,
44+
parent_task_external_id,
45+
idempotency_key
46+
FROM new_rows
47+
WHERE dag_id IS NULL
48+
ON CONFLICT (inserted_at, id) DO NOTHING;
49+
50+
INSERT INTO v1_lookup_table_olap (
51+
tenant_id,
52+
external_id,
53+
task_id,
54+
inserted_at
55+
)
56+
SELECT
57+
tenant_id,
58+
external_id,
59+
id,
60+
inserted_at
61+
FROM new_rows
62+
ON CONFLICT (external_id) DO NOTHING;
63+
64+
-- If the task has a dag_id and dag_inserted_at, insert into the lookup table
65+
INSERT INTO v1_dag_to_task_olap (
66+
dag_id,
67+
dag_inserted_at,
68+
task_id,
69+
task_inserted_at
70+
)
71+
SELECT
72+
dag_id,
73+
dag_inserted_at,
74+
id,
75+
inserted_at
76+
FROM new_rows
77+
WHERE dag_id IS NOT NULL
78+
ON CONFLICT (dag_id, dag_inserted_at, task_id, task_inserted_at) DO NOTHING;
79+
80+
RETURN NULL;
81+
END;
82+
$$
83+
LANGUAGE plpgsql;
84+
85+
CREATE OR REPLACE FUNCTION v1_dags_olap_insert_function()
86+
RETURNS TRIGGER AS
87+
$$
88+
BEGIN
89+
INSERT INTO v1_runs_olap (
90+
tenant_id,
91+
id,
92+
inserted_at,
93+
external_id,
94+
readable_status,
95+
kind,
96+
workflow_id,
97+
workflow_version_id,
98+
additional_metadata,
99+
parent_task_external_id,
100+
idempotency_key
101+
)
102+
SELECT
103+
tenant_id,
104+
id,
105+
inserted_at,
106+
external_id,
107+
readable_status,
108+
'DAG',
109+
workflow_id,
110+
workflow_version_id,
111+
additional_metadata,
112+
parent_task_external_id,
113+
idempotency_key
114+
FROM new_rows
115+
ON CONFLICT (inserted_at, id) DO NOTHING;
116+
117+
INSERT INTO v1_lookup_table_olap (
118+
tenant_id,
119+
external_id,
120+
dag_id,
121+
inserted_at
122+
)
123+
SELECT
124+
tenant_id,
125+
external_id,
126+
id,
127+
inserted_at
128+
FROM new_rows
129+
ON CONFLICT (external_id) DO NOTHING;
130+
131+
RETURN NULL;
132+
END;
133+
$$
134+
LANGUAGE plpgsql;
135+
-- +goose StatementEnd
136+
137+
-- +goose Down
138+
-- +goose StatementBegin
139+
ALTER TABLE "WorkflowVersion"
140+
DROP COLUMN "idempotencyKeyExpression",
141+
DROP COLUMN "idempotencyKeyTtlMs"
142+
;
143+
144+
ALTER TABLE v1_task DROP COLUMN idempotency_key;
145+
ALTER TABLE v1_dag DROP COLUMN idempotency_key;
146+
147+
ALTER TABLE v1_dags_olap DROP COLUMN idempotency_key;
148+
ALTER TABLE v1_tasks_olap DROP COLUMN idempotency_key;
149+
ALTER TABLE v1_runs_olap DROP COLUMN idempotency_key;
150+
151+
CREATE OR REPLACE FUNCTION v1_tasks_olap_insert_function()
152+
RETURNS TRIGGER AS
153+
$$
154+
BEGIN
155+
INSERT INTO v1_runs_olap (
156+
tenant_id,
157+
id,
158+
inserted_at,
159+
external_id,
160+
readable_status,
161+
kind,
162+
workflow_id,
163+
workflow_version_id,
164+
additional_metadata,
165+
parent_task_external_id
166+
)
167+
SELECT
168+
tenant_id,
169+
id,
170+
inserted_at,
171+
external_id,
172+
readable_status,
173+
'TASK',
174+
workflow_id,
175+
workflow_version_id,
176+
additional_metadata,
177+
parent_task_external_id
178+
FROM new_rows
179+
WHERE dag_id IS NULL
180+
ON CONFLICT (inserted_at, id) DO NOTHING;
181+
182+
INSERT INTO v1_lookup_table_olap (
183+
tenant_id,
184+
external_id,
185+
task_id,
186+
inserted_at
187+
)
188+
SELECT
189+
tenant_id,
190+
external_id,
191+
id,
192+
inserted_at
193+
FROM new_rows
194+
ON CONFLICT (external_id) DO NOTHING;
195+
196+
-- If the task has a dag_id and dag_inserted_at, insert into the lookup table
197+
INSERT INTO v1_dag_to_task_olap (
198+
dag_id,
199+
dag_inserted_at,
200+
task_id,
201+
task_inserted_at
202+
)
203+
SELECT
204+
dag_id,
205+
dag_inserted_at,
206+
id,
207+
inserted_at
208+
FROM new_rows
209+
WHERE dag_id IS NOT NULL
210+
ON CONFLICT (dag_id, dag_inserted_at, task_id, task_inserted_at) DO NOTHING;
211+
212+
RETURN NULL;
213+
END;
214+
$$
215+
LANGUAGE plpgsql;
216+
217+
CREATE OR REPLACE FUNCTION v1_dags_olap_insert_function()
218+
RETURNS TRIGGER AS
219+
$$
220+
BEGIN
221+
INSERT INTO v1_runs_olap (
222+
tenant_id,
223+
id,
224+
inserted_at,
225+
external_id,
226+
readable_status,
227+
kind,
228+
workflow_id,
229+
workflow_version_id,
230+
additional_metadata,
231+
parent_task_external_id
232+
)
233+
SELECT
234+
tenant_id,
235+
id,
236+
inserted_at,
237+
external_id,
238+
readable_status,
239+
'DAG',
240+
workflow_id,
241+
workflow_version_id,
242+
additional_metadata,
243+
parent_task_external_id
244+
FROM new_rows
245+
ON CONFLICT (inserted_at, id) DO NOTHING;
246+
247+
INSERT INTO v1_lookup_table_olap (
248+
tenant_id,
249+
external_id,
250+
dag_id,
251+
inserted_at
252+
)
253+
SELECT
254+
tenant_id,
255+
external_id,
256+
id,
257+
inserted_at
258+
FROM new_rows
259+
ON CONFLICT (external_id) DO NOTHING;
260+
261+
RETURN NULL;
262+
END;
263+
$$
264+
LANGUAGE plpgsql;
265+
-- +goose StatementEnd

0 commit comments

Comments
 (0)