feat(engine): idempotency

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

@@ -121,6 +121,9 @@ V1TaskSummary:
       type: string
       description: The external ID of the parent task.
       format: uuid
+    idempotencyKey:
+      type: string
+      description: The idempotency key that was claimed by the task run
   required:
     - metadata
     - createdAt

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

@@ -119,6 +119,14 @@ listWorkflowRuns:
         required: false
         schema:
           $ref: "../../../components/schemas/_index.yaml#/V1RunningFilter"
+      - description: The idempotency key(s) to filter for
+        in: query
+        name: idempotency_keys
+        required: false
+        schema:
+          type: array
+          items:
+            type: string
     responses:
       "200":
         content:

api-contracts/v1/workflows.proto

@@ -111,8 +111,24 @@ message CreateWorkflowVersionRequest {
     repeated Concurrency concurrency_arr = 12; // (optional) the workflow concurrency options
     repeated DefaultFilter default_filters = 13; // (optional) the default filters for the workflow
     optional bytes input_json_schema = 14; // (optional) the JSON schema for the workflow input
+
+    optional IdempotencyConfig idempotency = 15; // (optional) idempotency configuration for the workflow
+}
+
+message IdempotencyConfig {
+    string expression = 1; // a CEL expression for determining the idempotency key for workflow runs
+    int64 ttl_ms = 2; // time-to-live for idempotency keys in milliseconds
 }
 
+message IdempotencyCollisionError {
+    string existing_run_external_id = 1; // the external ID of the existing workflow run that caused the collision
+    string colliding_run_external_id = 2; // the external ID of the workflow run that caused the collision
+}
+
+message BulkTriggerIdempotencyCollisionError {
+    repeated string successful_workflow_run_external_ids = 1; // the external IDs of the successfully triggered workflow runs
+    repeated IdempotencyCollisionError collisions = 2; // the idempotency collision errors
+}
 
 message DefaultFilter {
     string expression = 1; // (required) the CEL expression for the filter

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

@@ -140,6 +140,7 @@ func (t *V1WorkflowRunsService) WithDags(ctx context.Context, request gen.V1Work
 		Offset:          offset,
 		IncludePayloads: includePayloads,
 		UseGinIndex:     useGinIndex,
+		IdempotencyKeys: request.Params.IdempotencyKeys,
 	}
 
 	additionalMetadataFilters := make(map[string]interface{})
@@ -291,6 +292,7 @@ func (t *V1WorkflowRunsService) OnlyTasks(ctx context.Context, request gen.V1Wor
 		WorkerId:        request.Params.WorkerId,
 		IncludePayloads: includePayloads,
 		UseGinIndex:     useGinIndex,
+		IdempotencyKeys: request.Params.IdempotencyKeys,
 	}
 
 	additionalMetadataFilters := make(map[string]interface{})

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

@@ -77,7 +77,7 @@ func ToV1Event(event *v1.EventWithPayload) gen.V1Event {
 		},
 		Payload:               &payload,
 		SeenAt:                &event.EventSeenAt.Time,
-		Scope:                 &event.EventScope,
+		Scope:                 event.EventScope,
 		TriggeredRuns:         &triggeredRuns,
 		TriggeringWebhookName: event.TriggeringWebhookName,
 	}

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

@@ -105,6 +105,7 @@ func WorkflowRunDataToV1TaskSummary(task *v1.WorkflowRunData, workflowIdsToNames
 		RetryCount:            &retryCount,
 		Attempt:               &attempt,
 		ParentTaskExternalId:  parentTaskExternalId,
+		IdempotencyKey:        task.IdempotencyKey,
 	}
 
 	if isEvicted {
@@ -192,6 +193,12 @@ func PopulateTaskRunDataRowToV1TaskSummary(task *v1.TaskWithPayloads, workflowNa
 
 	taskStatus, isEvicted := mapOlapStatus(string(task.Status))
 
+	var idempotencyKey *string
+
+	if task.IdempotencyKey.Valid {
+		idempotencyKey = &task.IdempotencyKey.String
+	}
+
 	summary := gen.V1TaskSummary{
 		Metadata: gen.APIResourceMeta{
 			Id:        task.ExternalID.String(),
@@ -223,6 +230,7 @@ func PopulateTaskRunDataRowToV1TaskSummary(task *v1.TaskWithPayloads, workflowNa
 		Attempt:               &attempt,
 		WorkflowRunExternalId: task.WorkflowRunID,
 		ParentTaskExternalId:  task.ParentTaskExternalID,
+		IdempotencyKey:        idempotencyKey,
 	}
 
 	if isEvicted {

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

@@ -0,0 +1,265 @@
+-- +goose Up
+-- +goose StatementBegin
+ALTER TABLE "WorkflowVersion"
+    ADD COLUMN "idempotencyKeyExpression" TEXT,
+    ADD COLUMN "idempotencyKeyTtlMs" BIGINT
+    ;
+
+ALTER TYPE v1_cel_evaluation_failure_source ADD VALUE IF NOT EXISTS 'IDEMPOTENCY_KEY';
+
+ALTER TABLE v1_task ADD COLUMN idempotency_key TEXT;
+ALTER TABLE v1_dag ADD COLUMN idempotency_key TEXT;
+
+ALTER TABLE v1_tasks_olap ADD COLUMN idempotency_key TEXT;
+ALTER TABLE v1_dags_olap ADD COLUMN idempotency_key TEXT;
+ALTER TABLE v1_runs_olap ADD COLUMN idempotency_key TEXT;
+
+CREATE OR REPLACE FUNCTION v1_tasks_olap_insert_function()
+RETURNS TRIGGER AS
+$$
+BEGIN
+    INSERT INTO v1_runs_olap (
+        tenant_id,
+        id,
+        inserted_at,
+        external_id,
+        readable_status,
+        kind,
+        workflow_id,
+        workflow_version_id,
+        additional_metadata,
+        parent_task_external_id,
+        idempotency_key
+    )
+    SELECT
+        tenant_id,
+        id,
+        inserted_at,
+        external_id,
+        readable_status,
+        'TASK',
+        workflow_id,
+        workflow_version_id,
+        additional_metadata,
+        parent_task_external_id,
+        idempotency_key
+    FROM new_rows
+    WHERE dag_id IS NULL
+    ON CONFLICT (inserted_at, id) DO NOTHING;
+
+    INSERT INTO v1_lookup_table_olap (
+        tenant_id,
+        external_id,
+        task_id,
+        inserted_at
+    )
+    SELECT
+        tenant_id,
+        external_id,
+        id,
+        inserted_at
+    FROM new_rows
+    ON CONFLICT (external_id) DO NOTHING;
+
+    -- If the task has a dag_id and dag_inserted_at, insert into the lookup table
+    INSERT INTO v1_dag_to_task_olap (
+        dag_id,
+        dag_inserted_at,
+        task_id,
+        task_inserted_at
+    )
+    SELECT
+        dag_id,
+        dag_inserted_at,
+        id,
+        inserted_at
+    FROM new_rows
+    WHERE dag_id IS NOT NULL
+    ON CONFLICT (dag_id, dag_inserted_at, task_id, task_inserted_at) DO NOTHING;
+
+    RETURN NULL;
+END;
+$$
+LANGUAGE plpgsql;
+
+CREATE OR REPLACE FUNCTION v1_dags_olap_insert_function()
+RETURNS TRIGGER AS
+$$
+BEGIN
+    INSERT INTO v1_runs_olap (
+        tenant_id,
+        id,
+        inserted_at,
+        external_id,
+        readable_status,
+        kind,
+        workflow_id,
+        workflow_version_id,
+        additional_metadata,
+        parent_task_external_id,
+        idempotency_key
+    )
+    SELECT
+        tenant_id,
+        id,
+        inserted_at,
+        external_id,
+        readable_status,
+        'DAG',
+        workflow_id,
+        workflow_version_id,
+        additional_metadata,
+        parent_task_external_id,
+        idempotency_key
+    FROM new_rows
+    ON CONFLICT (inserted_at, id) DO NOTHING;
+
+    INSERT INTO v1_lookup_table_olap (
+        tenant_id,
+        external_id,
+        dag_id,
+        inserted_at
+    )
+    SELECT
+        tenant_id,
+        external_id,
+        id,
+        inserted_at
+    FROM new_rows
+    ON CONFLICT (external_id) DO NOTHING;
+
+    RETURN NULL;
+END;
+$$
+LANGUAGE plpgsql;
+-- +goose StatementEnd
+
+-- +goose Down
+-- +goose StatementBegin
+ALTER TABLE "WorkflowVersion"
+    DROP COLUMN "idempotencyKeyExpression",
+    DROP COLUMN "idempotencyKeyTtlMs"
+;
+
+ALTER TABLE v1_task DROP COLUMN idempotency_key;
+ALTER TABLE v1_dag DROP COLUMN idempotency_key;
+
+ALTER TABLE v1_dags_olap DROP COLUMN idempotency_key;
+ALTER TABLE v1_tasks_olap DROP COLUMN idempotency_key;
+ALTER TABLE v1_runs_olap DROP COLUMN idempotency_key;
+
+CREATE OR REPLACE FUNCTION v1_tasks_olap_insert_function()
+RETURNS TRIGGER AS
+$$
+BEGIN
+    INSERT INTO v1_runs_olap (
+        tenant_id,
+        id,
+        inserted_at,
+        external_id,
+        readable_status,
+        kind,
+        workflow_id,
+        workflow_version_id,
+        additional_metadata,
+        parent_task_external_id
+    )
+    SELECT
+        tenant_id,
+        id,
+        inserted_at,
+        external_id,
+        readable_status,
+        'TASK',
+        workflow_id,
+        workflow_version_id,
+        additional_metadata,
+        parent_task_external_id
+    FROM new_rows
+    WHERE dag_id IS NULL
+    ON CONFLICT (inserted_at, id) DO NOTHING;
+
+    INSERT INTO v1_lookup_table_olap (
+        tenant_id,
+        external_id,
+        task_id,
+        inserted_at
+    )
+    SELECT
+        tenant_id,
+        external_id,
+        id,
+        inserted_at
+    FROM new_rows
+    ON CONFLICT (external_id) DO NOTHING;
+
+    -- If the task has a dag_id and dag_inserted_at, insert into the lookup table
+    INSERT INTO v1_dag_to_task_olap (
+        dag_id,
+        dag_inserted_at,
+        task_id,
+        task_inserted_at
+    )
+    SELECT
+        dag_id,
+        dag_inserted_at,
+        id,
+        inserted_at
+    FROM new_rows
+    WHERE dag_id IS NOT NULL
+    ON CONFLICT (dag_id, dag_inserted_at, task_id, task_inserted_at) DO NOTHING;
+
+    RETURN NULL;
+END;
+$$
+LANGUAGE plpgsql;
+
+CREATE OR REPLACE FUNCTION v1_dags_olap_insert_function()
+RETURNS TRIGGER AS
+$$
+BEGIN
+    INSERT INTO v1_runs_olap (
+        tenant_id,
+        id,
+        inserted_at,
+        external_id,
+        readable_status,
+        kind,
+        workflow_id,
+        workflow_version_id,
+        additional_metadata,
+        parent_task_external_id
+    )
+    SELECT
+        tenant_id,
+        id,
+        inserted_at,
+        external_id,
+        readable_status,
+        'DAG',
+        workflow_id,
+        workflow_version_id,
+        additional_metadata,
+        parent_task_external_id
+    FROM new_rows
+    ON CONFLICT (inserted_at, id) DO NOTHING;
+
+    INSERT INTO v1_lookup_table_olap (
+        tenant_id,
+        external_id,
+        dag_id,
+        inserted_at
+    )
+    SELECT
+        tenant_id,
+        external_id,
+        id,
+        inserted_at
+    FROM new_rows
+    ON CONFLICT (external_id) DO NOTHING;
+
+    RETURN NULL;
+END;
+$$
+LANGUAGE plpgsql;
+-- +goose StatementEnd