Idempotency

Understand what Flux guarantees through durable replay — completed tasks are never re-executed — and how to design task side effects that stay safe when a task runs more than once.

Flux guarantees that a task whose result was persisted will never run again on replay — that’s the core of the durable replay model. The narrow case to plan for is a task that ran most of the way through but didn’t finish persisting before the process went down. The patterns on this page make those tasks safe to retry, so the end-to-end behavior is exactly-once from the perspective of the external systems your workflow touches.

What Flux records and replays

Every task that completes successfully emits a TASK_COMPLETED event. The event carries the task’s output and a stable task_id derived from the task’s name and its input arguments. When a workflow is resumed — after a crash, a paused state, or an operator-initiated replay — Flux scans the event log for each task as it would naturally be called. If a matching TASK_COMPLETED event exists, Flux returns the recorded output immediately and never enters the task body:

from flux import ExecutionContext
from flux.task import task
from flux.workflow import workflow

call_count = 0

@task
async def fetch_data(record_id: str) -> dict:
    global call_count
    call_count += 1
    return {"id": record_id, "value": 99}

@workflow
async def process(ctx: ExecutionContext[str]):
    result = await fetch_data(ctx.input)
    return result

if __name__ == "__main__":
    ctx = process.run("rec-001")
    print(f"First run — function called: {call_count} time(s)")  # 1

    ctx2 = process.resume(ctx.execution_id)
    print(f"After resume — function called: {call_count} time(s)")  # still 1
    print(ctx2.output)  # {'id': 'rec-001', 'value': 99}

The task function is called exactly once. On resume, Flux replays past the completed task and returns the same output without touching the network, database, or any other external system.

The identity of a task invocation

The lookup key — the task_id — is a hash of the task’s name and its arguments. Two calls to the same function with the same arguments produce the same task_id. Two calls with different arguments produce different task_id values, so Flux treats them as distinct invocations and records each one separately.

This is why the task name and arguments must be deterministic. If arguments include timestamps, random numbers, or UUIDs generated outside a task, each replay produces a different task_id and Flux finds no matching event — it re-executes the task body as if it were running for the first time. Keep non-determinism inside tasks, not in the arguments passed to them. The Concepts: determinism page covers this constraint in detail.

When a task may run more than once

The replay guarantee starts at the task boundary: Flux records a TASK_COMPLETED event after the task function returns and ctx.checkpoint() runs. If a worker dies between the return and the checkpoint, the next run sees no completion event and re-enters the task body. This window is narrow, but in a long-running production system it’s a window that will eventually open. Designing tasks to be safe under re-execution turns that case into a non-event. Two patterns cover almost everything you’ll write.

Strategy 1: pass an idempotency key to the external API

Most payment processors, messaging platforms, and cloud-provisioning APIs accept an idempotency key — a caller-supplied string that tells the API “if you have already processed a request with this key, return the same result without repeating the operation.” Constructing the key from the execution context ties it to a specific, unrepeatable task invocation:

from flux import ExecutionContext
from flux.task import task
from flux.workflow import workflow

@task
async def charge(user_id: str, amount: float) -> dict:
    ctx = await ExecutionContext.get()
    # Combine execution_id with the distinguishing arguments so the key
    # is unique per execution and per (user_id, amount) pair.
    key = f"charge_{ctx.execution_id}_{user_id}"
    return await payment_api.charge(
        user_id=user_id,
        amount=amount,
        idempotency_key=key,
    )

@workflow
async def checkout(ctx: ExecutionContext):
    return await charge("user-42", 99.99)

If the task runs again before TASK_COMPLETED is checkpointed, the API receives the same key and returns the cached result. The user is charged once.

Strategy 2: check before you act

When the external system does not support idempotency keys, use a read-before-write pattern: check whether the resource already exists (or the action already happened) before performing the write. Return early if the state is already correct:

from flux import ExecutionContext
from flux.task import task
from flux.workflow import workflow

@task
async def ensure_record(record_id: str, data: dict) -> dict:
    """Create the record only if it does not already exist."""
    existing = await db.find(record_id)
    if existing:
        return existing  # already done — return stored result
    return await db.create(record_id, data)

@workflow
async def provision(ctx: ExecutionContext):
    return await ensure_record("user-42", {"name": "Alice", "plan": "pro"})

The task is now safe regardless of how many times it runs. The second call returns the same object as the first.

Unsafe patterns and how to fix them

Unconditional append

Writing to a log, queue, or ledger without checking for duplicates means each re-execution adds another entry:

# Unsafe — appends on every call
@task
async def log_event(event_id: str) -> None:
    await audit_log.append(event_id)

# Safe — checks first, or passes an idempotency key
@task
async def log_event(event_id: str) -> None:
    if not await audit_log.exists(event_id):
        await audit_log.append(event_id)

Sending without deduplication

Emails, SMS, and webhook calls sent unconditionally duplicate on retry. Fix this by tracking the send in a store keyed by the task’s idempotency key, or by using a delivery system that accepts a deduplication ID.

Generating a unique ID outside a task

If the workflow generates a UUID and passes it into a task, each replay generates a different UUID and produces a different task_id. Flux sees no prior completion and re-runs the task. Generate IDs inside tasks using flux.tasks.uuid4() — a built-in task that records its output and replays the same value:

from flux.tasks import uuid4
from flux import ExecutionContext
from flux.workflow import workflow

@workflow
async def create_order(ctx: ExecutionContext):
    order_id = await uuid4()  # same value on every replay
    return await provision_order(order_id)

Compensation tasks need idempotency too

The Rollback and compensation page notes that compensation tasks can run more than once if the workflow is interrupted mid-compensation. Apply the same strategies: check before undoing, or pass the idempotency key to the compensation API. A compensation that charges a refund unconditionally on each call is exactly as dangerous as a charge that does the same.