Great question! In Prefect 3.x, both @task and @flow accept the same retry-related parameters. Here’s how

retries

and

retry_delay_seconds

work and how to use them effectively. What they do - retries - Number of times Prefect will retry after a failure. - Total possible attempts = 1 initial attempt + retries. - Default: no retries if not set or set to 0. - retry_delay_seconds - The delay before each retry, in seconds. - Applies to tasks and flows independently (see interaction notes below). - If not provided, retries are scheduled immediately (no delay). Accepted values for retry_delay_seconds - Single number (int/float) - Use the same delay before every retry. - Example:

retry_delay_seconds=10

delays 10s before each retry. - List of numbers - Use a different delay per retry in order. - The list should provide one delay per retry; keep the length equal to

retries

. - Example:

retry_delay_seconds=[5, 15, 30]

with

retries=3

. - Callable returning a list[float] - Generate a full schedule programmatically based on the number of retries. - Prefect provides

exponential_backoff(backoff_factor)

for convenience. - Example:

retry_delay_seconds=exponential_backoff(5)

will generate a list of delays for the configured

retries

. Jitter (optional but useful) - retry_jitter_factor - Adds randomness to each delay to avoid “thundering herd” issues. - Example:

retry_jitter_factor=0.2

will add ±20% jitter to each delay. Flow vs Task retries - They are independent: - Task retries: Only the failed task is retried according to its task-level settings. - Flow retries: The entire flow is retried (re-run from the start) if the flow run fails. - If a task exhausts its retries and causes the flow to fail, and the flow has its own retries, the entire flow will run again — and that task will get its full set of retries again on the new flow attempt. Examples - Fixed delay for a task

from prefect import task, flow

@task(retries=3, retry_delay_seconds=10)  # 3 retries, 10s between each
def flaky_task():
    ...

@flow
def my_flow():
    flaky_task()

- Per-attempt delays (list)

from prefect import task

@task(retries=3, retry_delay_seconds=[5, 10, 30])  # 5s, then 10s, then 30s
def flaky_task():
    ...

- Generated schedule with exponential backoff

from prefect import task
from prefect.tasks import exponential_backoff

@task(
    retries=4,
    retry_delay_seconds=exponential_backoff(5),  # base backoff factor (seconds)
    retry_jitter_factor=0.2,                     # ±20% jitter
)
def flaky_task():
    ...

- Custom callable schedule

from prefect import task

# Create a capped exponential backoff schedule (e.g., 1, 2, 4, 8, capped at 60s)
@task(
    retries=4,
    retry_delay_seconds=lambda n: [min(2 ** i, 60) for i in range(n)]
)
def flaky_task():
    ...

- Flow-level retries

from prefect import flow, task

@task(retries=2, retry_delay_seconds=15)
def sometimes_fails():
    ...

@flow(retries=2, retry_delay_seconds=60)  # rerun the whole flow after 60s if it fails
def pipeline():
    sometimes_fails()

Docs - How-to guide: Retries in Prefect 3 - Concepts: Retry schedules and delays If you share your desired retry pattern (e.g., fixed, exponential with cap, custom), I can help write the exact decorator configuration.