Short answer: make sure your flow “touches” the futures from the mapped/concurrent tasks. In Prefect 3.x, a flow only fails when an exception escapes the flow function. Calling

.result()

(or resolving futures) will raise if any task failed, which marks the flow run as Failed. Two common patterns: 1) Fail fast (on the first failing element)

from prefect import flow, task
from prefect.futures import as_completed

@task
def work(x: int) -> int:
    if x == 3:
        raise ValueError("boom")
    return x * 2

@flow
def main(items: list[int]):
    futures = [work.submit(i) for i in items]

    # As soon as any future fails, this raises and the flow fails
    for fut in as_completed(futures):
        fut.result()

    # If we got here, everything succeeded
    return [f.result() for f in futures]

2) Let everything finish, then fail if any failed (aggregate check)

from prefect import flow, task
from prefect.futures import resolve_futures_to_states
from prefect.states import raise_state_exception

@task
def work(x: int) -> int:
    if x == 3:
        raise ValueError("boom")
    return x * 2

@flow
def main(items: list[int]):
    futures = [work.submit(i) for i in items]

    # Collect states for all futures
    states = resolve_futures_to_states(futures)

    # If any failed, raise one of the failures to fail the flow
    failed_states = [s for s in states if s.is_failed()]
    if failed_states:
        raise_state_exception(failed_states[0])

    # Otherwise, return the successful results
    return [f.result() for f in futures]

Notes: - If you used

prefect.allow_failure(...)

upstream, that intentionally prevents failures from failing downstream work. Remove it if you want the flow to fail when any element fails, or keep it and do the explicit post-check (pattern 2). - You can also resolve structures of futures in one go and let failures raise:

from prefect.futures import resolve_futures_to_results
  results = resolve_futures_to_results(futures)  # raises if any element failed

References: - Conceptual background on mapping/concurrency: Beyond loops: how Prefect’s task mapping scales - Source for flow behavior: prefect/flows.py If you can share a small snippet of how you’re “mapping” today (e.g., using `submit`/`as_completed`/`allow_failure`), I can tailor the example to your case.