Mapped children not yet started leave no task run when PrefectFutureList.result() raises the first failure
Bug summary
When PrefectFutureList.result() raises the first failure it sees, mapped children that have not started yet are dropped by the task runner's shutdown(cancel_futures=True) and leave no task run at all. The flow run shows fewer children than were mapped, with no CANCELLED state to explain the gap.
import time
import prefect
from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner
@task
def work(x):
if x == 0:
raise ValueError("first fails")
time.sleep(2)
return x
@flow(task_runner=ThreadPoolTaskRunner(max_workers=1))
def flow_a():
return work.map([0, 1, 2]).result()
@flow(task_runner=ThreadPoolTaskRunner(max_workers=1))
def flow_b():
futures = work.map([0, 1, 2])
prefect.futures.wait(futures)
return [f.result(raise_on_failure=False) for f in futures]Task runs read back through the client after each run:
flow_a final state: FAILED - Flow run encountered an exception: ValueError: first fails
work-1da FAILED Task run encountered an exception ValueError: first fails
work-f83 COMPLETED
<- the third mapped call has no task run
flow_b final state: COMPLETED
work-885 COMPLETED
work-94a FAILED Task run encountered an exception ValueError: first fails
work-bdc COMPLETEDDiscussion
The early raise itself is documented in the result() docstring ("Uses as_completed internally so that failures are raised as soon as they occur"). What is not visible to the user is what happens to the siblings: ThreadPoolTaskRunner.__exit__ (and ProcessPoolTaskRunner.__exit__) call shutdown(cancel_futures=True) unconditionally, so anything still queued is discarded before its task run is created. With ~50 mapped tables in production, one bad table stopped the other unstarted ones from running, and the run gave no sign of which ones were skipped. On 3.8.4 we also saw ValueError: Expected failed or crashed state got NotReady(...) surface from .result() in that situation, naming neither the failing child nor the real error; I could not reproduce that message on 3.8.6, so I only note it.
flow_b is the pattern the docs give for inspecting every outcome, and it behaves well, but there is no built-in that waits for all mapped children and then raises naming every failure.
Suggested change
One or both of:
- record queued children discarded by the runner shutdown as
CANCELLEDtask runs (or log them), so the run shows what did not execute; - give
PrefectFutureList.result()a way to wait for every future before raising (for example collecting the failures and raising one exception that names them), so a mapped call over many items is not cut short by its first failure.
Version info
Version: 3.8.6
Python version: 3.12.13
Server type: ephemeral
Pydantic version: 2.13.5Source: PrefectHQ/prefect