Errors and retries

Any task that talks to a network, a database, or another process will eventually hit a transient failure. Flux gives you three options on task.with_options() to handle these failures at the task boundary, before they propagate to the workflow: retry_max_attempts, retry_delay, and fallback.

How retries work

When a task raises an unhandled exception, Flux checks whether retry_max_attempts > 0. If it is, Flux waits retry_delay seconds, then calls the function again. This repeats up to retry_max_attempts times. If every attempt fails, Flux raises an ExecutionError wrapping the last exception — or calls the fallback function if one is configured.

retry_max_attempts counts attempts beyond the first, not total attempts. A task with retry_max_attempts=2 can run at most three times: one initial attempt plus two retries.

This task retries twice on failure. It fails on the first two calls and succeeds on the third:

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

attempt_count = 0


@task.with_options(
    retry_max_attempts=2,
    retry_delay=1,
)
async def fetch_record(record_id: str) -> dict:
    global attempt_count
    attempt_count += 1
    if attempt_count < 3:
        raise ConnectionError("Transient network failure")
    return {"id": record_id, "value": 42}


@workflow
async def load_record(ctx: ExecutionContext[str]):
    return await fetch_record(ctx.input)


if __name__ == "__main__":
    ctx = load_record.run("rec-001")
    print(ctx.output)         # {'id': 'rec-001', 'value': 42}
    print(ctx.has_succeeded)  # True
    print(attempt_count)      # 3

Flux records the task’s output after the first successful attempt. On a workflow replay, it returns the recorded value without re-running the function — retries are a concern of the first execution only.

The retry_delay option

retry_delay sets the base wait, in seconds, before the first retry. The default is 1. Subsequent retries scale this value by retry_backoff (see below).

Set retry_delay to a value that reflects how long the transient condition typically lasts. A momentary load spike may need one or two seconds; a service restart or a rate-limit window may need ten to thirty.

The retry_backoff option

retry_backoff is a multiplier applied to the delay between successive retries — it produces genuine exponential backoff. The wait before the first retry is retry_delay; each subsequent retry multiplies the previous wait by retry_backoff. The delay is capped at 600 seconds.

With retry_delay=2 and retry_backoff=3, the delay sequence is 2 s → 6 s → 18 s. The default retry_backoff is 2, so with retry_delay=1 the sequence is 1 s → 2 s → 4 s → 8 s.

You can observe this in the event log: the current_delay field in each TASK_RETRY_STARTED event shows the compounding value used for that attempt.

The fallback option

fallback is a callable that Flux calls when all retries are exhausted. It receives the same arguments as the original task function. Its return value is recorded as the task’s output and the workflow continues normally — ctx.has_succeeded remains True.

Use fallback when the task’s failure is recoverable at the workflow level: return a default value, a cached result, or a sentinel that downstream tasks know how to handle.

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


def lookup_from_cache(record_id: str) -> dict:
    """Return cached data when the primary source is unreachable."""
    return {"id": record_id, "value": None, "source": "cache"}


@task.with_options(
    retry_max_attempts=2,
    retry_delay=1,
    fallback=lookup_from_cache,
)
async def fetch_record(record_id: str) -> dict:
    raise ConnectionError("Service unavailable")


@workflow
async def load_record(ctx: ExecutionContext[str]):
    return await fetch_record(ctx.input)


if __name__ == "__main__":
    ctx = load_record.run("rec-007")
    print(ctx.output)         # {'id': 'rec-007', 'value': None, 'source': 'cache'}
    print(ctx.has_succeeded)  # True

The fallback is called once. It is not retried. If the fallback itself raises, Flux emits a TASK_FALLBACK_FAILED event and fails the workflow. Wrap the fallback body in a try/except if you need it to be unconditionally safe.

Combining retry_max_attempts, retry_delay, and fallback

The two mechanisms compose. Flux exhausts all retries before calling the fallback:

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


def score_unavailable(user_id: str) -> dict:
    return {"user_id": user_id, "score": 0, "reliable": False}


@task.with_options(
    retry_max_attempts=3,
    retry_delay=2,
    fallback=score_unavailable,
)
async def fetch_credit_score(user_id: str) -> dict:
    raise TimeoutError("scoring service timed out")


@workflow
async def credit_check(ctx: ExecutionContext[str]):
    return await fetch_credit_score(ctx.input)


if __name__ == "__main__":
    ctx = credit_check.run("user-42")
    print(ctx.output)
    # {'user_id': 'user-42', 'score': 0, 'reliable': False}
    print(ctx.has_succeeded)  # True
    # fetch_credit_score was called 4 times total: 1 initial + 3 retries

The error-handling chain is: retry → fallback → rollback. If retry_max_attempts is set, Flux retries first. Only after all retries fail does Flux try the fallback. If there is no fallback, Flux calls rollback (if configured) and then raises. You can combine all three.

What to read next

• Rollback and compensation — how to undo side effects when a task fails permanently after all retries.
• Cancellation — cancelling a running workflow and the cleanup behavior that follows.