Add to Cursor Add to Claude Copy for LLM View as MD

Idempotency

If you need to prevent more than one run of a task from occurring within a given time window, for instance because of duplicate event sends from a webhook that can trigger duplicate runs in Hatchet, you can achieve this by adding idempotency configuration to your workflow or standalone task.

Define an Idempotency Key

Configuring idempotency on a workflow or standalone task requires two parameters to be set: the expression, which will be used to create an idempotency key from the input and additional metadata of the run that’s going to be triggered, and the TTL, which determines how long the key should live for.

⚠️

The idempotency key expression must evaluate to a string.

examples/python/idempotency/worker.py

EVENT_KEY = "idempotency:example"


class IdempotencyInput(BaseModel):
    id: str


@hatchet.task(
    idempotency=TTLBasedIdempotencyConfig(
        key_expression="input.id", ttl=timedelta(minutes=1)
    ),
    input_validator=IdempotencyInput,
    on_events=[EVENT_KEY],
)
async def idempotent_task(input: IdempotencyInput, ctx: Context) -> dict[str, str]:
    return {"result": f"Hello, world from task {input.id}"}

examples/typescript/idempotency/workflow.ts

export const idempotentTask = hatchet.task<IdempotencyInput, { result: string }>({
  name: 'ts-e2e-idempotent-task',
  idempotency: {
    strategy: 'ttl',
    expression: 'input.id',
    ttlMs: 60_000,
  },
  onEvents: [EVENT_KEY],
  fn: async (input) => {
    return { result: `Hello, world from task ${input.id}` };
  },
});

examples/go/idempotency/worker.go

func IdempotentTask(client *hatchet.Client) *hatchet.StandaloneTask {
	return client.NewStandaloneTask(
		"idempotent-task",
		func(ctx hatchet.Context, input IdempotencyInput) (*IdempotencyOutput, error) {
			return &IdempotencyOutput{
				Result: fmt.Sprintf("Hello, world from task %s", input.ID),
			}, nil
		},
		hatchet.WithWorkflowIdempotency(hatchet.IdempotencyConfig{
			Expression: "input.id",
			TTL:        time.Minute,
		}),
	)
}

examples/ruby/idempotency/worker.rb

IDEMPOTENT_TASK = HATCHET.task(
  name: 'ruby-e2e-idempotent-task',
  idempotency: Hatchet::TTLBasedIdempotencyConfig.new(expression: 'input.id', ttl_ms: 60_000),
  on_events: [EVENT_KEY]
) do |input, _ctx|
  { 'result' => "Hello from task #{input['id']}" }
end

IDEMPOTENT_TASK_SHORT_WINDOW = HATCHET.task(
  name: 'ruby-e2e-idempotent-task-short-window',
  idempotency: Hatchet::TTLBasedIdempotencyConfig.new(expression: 'input.id', ttl_ms: 2_000)
) do |input, _ctx|
  { 'result' => "Hello from task #{input['id']}" }
end

When idempotency is configured, only one run of a workflow will occur in the time window from when the first trigger comes in until the TTL expires. For instance, if you trigger a workflow at 00:00:00 UTC (midnight) with a TTL of five minutes, and then the same workflow is triggered again with the same inputs and metadata at 00:02:00 UTC, 00:04:00 UTC, and 00:06:00 UTC, only the first one (at midnight) and the final one (at 00:06:00 UTC) will run, and then after the second run occurs, the lock on the key will be held until 00:11:00 UTC (five minutes after the final run was triggered).

Handling Collisions

When a collision occurs, the engine will reject the workflow run, and, if the run was triggered from an SDK, then the SDK will raise an exception indicating that there was an idempotency collision. This exception will contain the id of the workflow run that already existed that had claimed the key already, so you can retrieve its output if you like.

examples/python/idempotency/trigger.py

ref_1 = await idempotent_task.aio_run(
    input=IdempotencyInput(id="123"),
    wait_for_result=False,
)

try:
    ref_2 = await idempotent_task.aio_run(
        input=IdempotencyInput(id="123"),
        wait_for_result=False,
    )
    run_id_2 = ref_2.workflow_run_id
except IdempotencyCollisionError as e:
    print(
        f"Run with external ID {e.existing_run_external_id} already exists for this idempotency key"
    )
    run_id_2 = e.existing_run_external_id

res_1 = await ref_1.aio_result()
res_2 = await idempotent_task.aio_get_result(run_id_2)

assert res_1 == res_2
assert ref_1.workflow_run_id == run_id_2

examples/typescript/idempotency/run.ts

const ref1 = await idempotentTask.runNoWait({ id: '123' });

let runId2: string;
try {
  const ref2 = await idempotentTask.runNoWait({ id: '123' });
  runId2 = await ref2.getWorkflowRunId();
} catch (e) {
  if (e instanceof IdempotencyCollisionError) {
    console.log(
      `Run with external ID ${e.existingRunExternalId} already exists for this idempotency key`
    );
    runId2 = e.existingRunExternalId;
  } else {
    throw e;
  }
}

const res1 = await ref1.result();
console.log(`Result: ${JSON.stringify(res1)}, run ID: ${runId2}`);

examples/go/idempotency/trigger.go

ref1, err := idempotentTask.RunNoWait(ctx, IdempotencyInput{ID: "123"})
if err != nil {
	log.Fatalf("unexpected error on first run: %v", err)
}

ref2, err := idempotentTask.RunNoWait(ctx, IdempotencyInput{ID: "123"})

var runID2 string

if err != nil {
	if idempErr, ok := hatchet.IsIdempotencyCollisionError(err); ok {
		fmt.Printf("Run %s already exists for this idempotency key\n", idempErr.ExistingRunExternalId)
		runID2 = idempErr.ExistingRunExternalId
	} else {
		log.Fatalf("unexpected error on second run: %v", err)
	}
} else {
	runID2 = ref2.RunId
}

examples/ruby/idempotency/trigger.rb

first_ref = IDEMPOTENT_TASK.run_no_wait({ 'id' => '123' })

second_run_id = begin
  second_ref = IDEMPOTENT_TASK.run_no_wait({ 'id' => '123' })
  second_ref.workflow_run_id
rescue Hatchet::IdempotencyCollisionError => e
  puts "Run #{e.existing_run_external_id} already exists for this idempotency key"
  e.existing_run_external_id
end

In other cases, such as triggering by events, the idempotency collision will be swallowed, and no additional runs will be created, but the event will still be ingested correctly without an error being raised.

Priority Durable Tasks

We use cookies

Idempotency

Define an Idempotency Key

Handling Collisions