Great question. Short version: -

flow.fn

lets you call the underlying Python function directly (no Prefect orchestration). - Prefect’s parallel mapping is on tasks, not flows. There’s no `flow.map(...)`; use

task.map(...)

or parallelize subflows via a wrapper or

flow.submit(...)

. Here’s how to combine these patterns for historical backfills. Key point about flow.fn - Calling

my_flow.fn(...)

bypasses Prefect entirely: no states, retries, caching, task runner, or UI/subflow tracking. - Because mapping is a task feature, you cannot map a flow directly. If you want Prefect-managed parallelism/observability, do not use

flow.fn

for the main work. Recommended pattern: factor work into a task and map it - Best for historical backfills over many partitions (e.g., dates) - Full Prefect observability, retries, and parallelism

from prefect import flow, task

@task
def process_date(date_str: str) -> str:
    # your historical processing for a single date partition
    return f"processed {date_str}"

@flow
def backfill(dates: list[str]) -> list[str]:
    futures = process_date.map(dates)  # runs in parallel via the flow's task runner
    return futures.result()

if __name__ == "__main__":
    backfill(["2025-01-01", "2025-01-02", "2025-01-03"])

Parallelize whole subflows (if you need per-date subflow runs in the UI) Option A: Wrap the subflow call in a task and map that - Keeps all subflow runs visible in the UI - Parallelism comes from mapping the wrapper task

from prefect import flow, task

@flow
def daily_flow(date_str: str) -> str:
    # this flow may contain tasks, retries, etc.
    return f"ran flow for {date_str}"

@task
def run_daily_subflow(date_str: str) -> str:
    # Calling a flow inside a task creates a subflow run tied to the parent flow
    return daily_flow(date_str)

@flow
def backfill(dates: list[str]) -> list[str]:
    futures = run_daily_subflow.map(dates)  # many subflows in parallel
    return futures.result()

Option B: Submit subflows directly (experimental) -

flow.submit(...)

queues independent flow runs (useful for large backfills on remote workers) - Marked experimental; API may change

from prefect import flow

@flow
def daily_flow(date_str: str) -> str:
    return f"ran flow for {date_str}"

@flow
def backfill(dates: list[str]) -> list[str]:
    futures = [daily_flow.submit(date_str=d) for d in dates]
    return [f.result() for f in futures]

Using flow.fn while still mapping (with caveats) - Only do this if your flow body is pure Python (no @task inside), or you don’t need inner-task observability/retries. - You’ll get parallelism from mapping the wrapper task, but anything inside the flow body will not run as Prefect tasks.

from prefect import flow, task

@flow
def daily_flow(date_str: str) -> str:
    # Pure-Python body only if you choose this route
    return f"computed {date_str}"

@task
def run_flow_body(date_str: str) -> str:
    # Calls the underlying function; not tracked as a subflow
    return daily_flow.fn(date_str)

@flow
def backfill(dates: list[str]) -> list[str]:
    futures = run_flow_body.map(dates)
    return futures.result()

Notes and tips - Concurrency is governed by the flow’s task runner. For CPU-heavy work, consider a process-based runner or running on workers in a work pool. For IO-bound work, the default concurrent runner is usually fine. - If you need to broadcast a static config to every mapped run, use

unmapped(...)

. - For very large backfills, consider creating a deployment and triggering many runs via the API/Automations for robust scaling. Docs for deeper details - Flows: Flows (Prefect 3) - Run work concurrently: How to run work concurrently - Tasks API: Tasks API Quick check: what’s your ideal outcome for the historical run?

- Do you want each date to appear as its own subflow with tracked tasks and retries? If yes, avoid

flow.fn

and use the subflow patterns above. - If you don’t need Prefect observability for the inner work and just want parallel execution, mapping a task that calls

flow.fn

is acceptable.

Short answer: returning

states.Failed(...)

from a plain helper function won’t change the flow’s state. Prefect only uses a returned State to set the run’s final state if that State is returned by the flow (or task) itself. In your snippet,

future_check

is just a regular function called inside the flow; the flow likely continues and returns something else (or nothing), so it ends as Completed. Why this happens: - Flows are marked Failed if they raise an exception, or if the flow function returns a manually constructed Failed state. - A task failing does not automatically fail the flow unless you propagate it (e.g., by calling

.result()

on the failed future which raises). - Returning

states.Failed

from a helper function has no effect unless the flow returns that state or raises. How to make the flow fail on your condition: Option A: Raise an exception in your helper (simplest)

from prefect import get_run_logger

def future_check(futures):
    logger = get_run_logger()
    if any(fut.state.is_failed() for fut in futures.values()):
        logger.error("One or more tasks failed. Please check the logs.")
        raise RuntimeError("One or more tasks failed. Please check the logs.")
    else:
        <http://logger.info|logger.info>("All tasks for file generation succeeded!")

Call it in the flow without catching the exception. Option B: Have the flow return the Failed state explicitly

from prefect import flow, get_run_logger, states

def future_check(futures):
    if any(fut.state.is_failed() for fut in futures.values()):
        return states.Failed(message="Flow completed with errors.")
    return None

@flow
def my_flow():
    # ... run tasks, collect `futures` dict ...
    state = future_check(futures)
    if state:
        return state  # This sets the flow run to Failed
    # otherwise continue and return your normal value

Option C: Propagate failures by touching results If you call

.result()

on any failed future inside the flow, it will raise the underlying exception and mark the flow as Failed:

for fut in futures.values():
    _ = fut.result()  # raises if task failed -> flow fails

If you want to check them all first (to log everything) and then fail the flow, use Option A or B after your inspection. Notes: - There is no

prefect.fail()

helper in Prefect 3.x; prefer raising exceptions or returning

states.Failed

from the flow itself. - Your logger line should be

<http://logger.info|logger.info>("All tasks for file generation succeeded!")

(Slack auto-linked it in your message). Docs for reference: - Prefect 3.x Concepts: States (how returning/raising affects run state) If you can share how you’re calling

future_check

inside the flow (and what the flow returns), I can suggest the minimal change for your exact code.

Sorry, I encountered an error while processing your request. Please try again.

Yes—if you return

states.Failed(...)

from the flow function, the flow run will be marked Failed. Two important caveats: - Make sure you’re checking the futures after they’ve finished; otherwise you might miss failures that haven’t materialized yet. You can block until completion with

wait

. - Logging alone doesn’t change state; you must either return a Failed state from the flow or raise an exception. Example:

from prefect import flow, get_run_logger, states
from prefect.futures import wait

@flow
def my_flow():
    logger = get_run_logger()
    # ... schedule tasks -> produce a list of PrefectFutures called all_futures_flat ...

    # ensure all futures are done before checking final states
    wait(all_futures_flat)

    if any(fut.state.is_failed() for fut in all_futures_flat):
        logger.error("One or more tasks failed. Please check the logs.")
        return states.Failed(message="Flow completed with errors.")
    else:
        <http://logger.info|logger.info>("All tasks for file generation succeeded!")
        return "ok"

Alternatively, you can fail the flow by raising:

if any(fut.state.is_failed() for fut in all_futures_flat):
    raise RuntimeError("One or more tasks failed. Please check the logs.")

Reference: Prefect 3.x Concepts: States

Short answer: no. A run that’s retrying is not “Cached(Completed)”. Here’s how it works in Prefect 3.x: - If a task attempt fails and there are retries left: - With a delay: the run moves to AwaitingRetry (Scheduled) - Without delay: the run moves to Retrying (Running) - It is not Completed or Cached during this period. - “Cached” is just a successful Completed state whose name is set to “Cached” because the result came from a cache hit. It only appears when a task returns a previously committed result, not when it fails. - A failed attempt never shows up as Completed or Cached. Only a successful run can be Completed/Cached. So even if a prior attempt failed, if a later retry hits cache or succeeds, that attempt will be Completed (possibly named “Cached”), but during the retry window the state is AwaitingRetry/Retrying, not Completed/Cached.