Timeouts

Unbounded tasks stall workflows. A downstream call that never returns will hold up every step that depends on its result. The timeout option on task.with_options() gives each task a hard deadline enforced by asyncio.wait_for. When the deadline passes, Flux cancels the task and raises ExecutionTimeoutError.

Setting a per-task timeout

Pass timeout=N (seconds, integer) in task.with_options(). Flux wraps the task body with asyncio.wait_for(coroutine, timeout=N). When the deadline expires, asyncio.wait_for raises Python’s built-in TimeoutError, which Flux catches and re-raises as ExecutionTimeoutError:

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

@task.with_options(timeout=5)
async def fetch_report(report_id: str) -> dict:
    # Simulates a slow external call
    await asyncio.sleep(30)
    return {"id": report_id, "data": "..."}

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

if __name__ == "__main__":
    ctx = generate_report.run("rpt-001")
    print(ctx.has_failed)   # True
    print(ctx.output)       # Task fetch_report (...) timed out (5s).

The timeout clock starts when the task function is entered — not when the task is scheduled or when its arguments are prepared.

The default is timeout=0, which means no deadline. Flux calls the function without asyncio.wait_for.

What ExecutionTimeoutError carries

ExecutionTimeoutError is a subclass of ExecutionError. It carries the same properties as ExecutionError, plus one of its own:

You can inspect it from the workflow body if you catch it explicitly:

from flux.errors import ExecutionTimeoutError

@workflow
async def generate_report(ctx: ExecutionContext[str]):
    try:
        return await fetch_report(ctx.input)
    except ExecutionTimeoutError as exc:
        print(f"deadline was {exc.timeout}s")
        raise  # re-raise so the workflow transitions to FAILED

Pairing timeout with a fallback

When a task times out and a fallback function is configured, Flux calls the fallback instead of failing the workflow. The fallback receives the same arguments as the original task. If the fallback returns normally, the workflow continues and ctx.has_succeeded stays True:

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

def cached_report(report_id: str) -> dict:
    """Return a stale cached result when the live source times out."""
    return {"id": report_id, "data": None, "source": "cache"}

@task.with_options(timeout=5, fallback=cached_report)
async def fetch_report(report_id: str) -> dict:
    await asyncio.sleep(30)
    return {"id": report_id, "data": "live"}

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

if __name__ == "__main__":
    ctx = generate_report.run("rpt-002")
    print(ctx.has_succeeded)  # True
    print(ctx.output)         # {'id': 'rpt-002', 'data': None, 'source': 'cache'}

The fallback is called once. If it raises, Flux emits a TASK_FALLBACK_FAILED event and fails the workflow. Wrap the fallback body in a try/except if it might fail in ways you need to handle gracefully.

Timeout with retries

When timeout and retry_max_attempts are both set, the timeout applies to every attempt, including each retry. Every attempt is wrapped in asyncio.wait_for(timeout=...), so a hung retry raises ExecutionTimeoutError instead of blocking indefinitely:

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

@task.with_options(timeout=1, retry_max_attempts=2, retry_delay=1)
async def unreliable_task() -> str:
    await asyncio.sleep(5)  # times out on every attempt, including retries
    return "done"

@workflow
async def demo(ctx: ExecutionContext):
    return await unreliable_task()

Each attempt of unreliable_task gets its own 1-second deadline. If an attempt times out, Flux retries (up to retry_max_attempts); if every attempt exceeds the deadline, the task raises ExecutionTimeoutError. The timeout counter does not carry over between attempts — each attempt starts with a fresh deadline.

Worker-side enforcement and blocking code

The Flux worker does not enforce per-task timeouts independently. timeout operates entirely inside the task’s asyncio coroutine via asyncio.wait_for — there is no server-side watchdog. If the task body is synchronous (blocking I/O, time.sleep, a CPU-bound loop), asyncio.wait_for cannot interrupt it: the coroutine must reach an await point for cancellation to land.

For blocking work, delegate to a thread pool with asyncio.to_thread:

import asyncio, time
from flux.task import task

@task.with_options(timeout=5)
async def cpu_bound_task(n: int) -> int:
    return await asyncio.to_thread(slow_computation, n)  # cancellable

def slow_computation(n: int) -> int:
    time.sleep(10)
    return n * n

What to read next

• Errors and retries — configure retry_max_attempts and fallback to recover from transient failures, including ones that trigger before a timeout.