I find reasoning about a Flow's final state confusing.

[ConstantinoSchillebeeckx]

I've been working on migrating our code from Prefect 1 -> Prefect 2 -> Prefect 3. I'm about done with the final migration however I'm left confused about knowing what the final state of my flow will be when one of it's constituent tasks fails. I'm hoping to start a discussion sharing what I'm experiencing as a user; at minimum, this will serve as my rubber duck and give me a guide I can reference in the future.

Below, I've shown numerous flows which should all (IMHO) terminate in a failed state since one of the tasks it has executed raises a ValueError. I should note that I'm new to the async world and so it's perfectly conceivable that I'm just not groking something. And yes, I've read this and this doc about terminal state.

I should also be explicit that my baseline expectation for the execution of flows are:

1. If any task fails in the flow, the final flow state should be failed (this of course assumes no use of raise_on_failure =False or any try/except code) - it's the flow's job to check error states, especially when a flow has a ton of tasks. This is why I generally don't return anything from my flow.
2. a flow should execute as many tasks as possible even if one fails; i.e. successful tasks should continue processing their downstream tasks

---

My flow examples can be found below, flow methods are named either:

• does_failX - when the final state of the flow is failed as expected ✅
• should_failX - when the final state of the flow is completed as NOT expected 😠

The tasks I'll be constructing my flows with are:

import asyncio
from prefect import flow, task

nums = [1, 2, 3]

@task
def foo(x):
    if x == 2:
        raise ValueError(x)
    return x

@task
async def bar(x):
    return x + 1

On to the flows ...

@flow(name="foo")
def does_fail1():
    foo(2)

Nice and simple, my flow does what I expect and fails because:

If an exception is raised directly in the flow function, the flow run is marked as FAILED.

@flow(name="foo")
def should_fail1():
    foo.submit(2)

Flow does not fail, I guess because the concurrent runner is executing it and doesn't raise the error "in the flow function"?

@flow(name="foo")
def does_fail2():
    return foo.submit(2)

Adding in a return does produce the expected state because:

If a flow returns a manually created state, it is used as the state of the final flow run. This allows for manual determination of final state.

I feel like this could produce some false-negative situations in cases where my flow has a ton of tasks and a developer forgot to return all needed tasks.

Let's now look at a slightly more complex set of flows.

@flow(name="foo")
def does_fail3():
    a = foo(2)
    foo(1, wait_for=[a])

Fails as expected, presumably because all the work happens inside the flow.

@flow(name="foo")
def should_fail2():
    a = foo.submit(2)
    foo(1, wait_for=[a])

Does not fail I guess for the same reason as should_fail1. We fixed that flow by adding a return statements, let's do that here.

@flow(name="foo")
def should_fail3():
    a = foo.submit(2)
    return foo(1, wait_for=[a])

Hmmm does not fail, why?

Let's now look at mapped versions of this flow.

@flow(name="foo")
async def should_fail4():
    foos = foo.map(nums)
    await asyncio.gather(*[bar(foo) for foo in foos])

Doesn't fail; let's add a return

@flow(name="foo")
async def should_fail5():
    foos = foo.map(nums)
    return await asyncio.gather(*[bar(foo) for foo in foos])

Hmm that didn't fail either. Why? Let's ask marvin how to fix it.

@flow(name="foo")
async def does_fail4():
    foos = foo.map(nums)
    return await asyncio.gather(*[bar.submit(foo) for foo in foos])

Technically does fail and my flow executes all the work I expected, but the failure reason is not what I expect:

 TypeError: An asyncio.Future, a coroutine or an awaitable is required

What if we use map instead of asyncio? I thought I had read docs that said submit/map don't have an async interface, so I wasn't aware I could try map. But @zzstoatzz suggested I try it:

@flow(name="foo")
async def should_fail6():
    foos = foo.map(nums)
    bar.map(foos)

No dice 😢

@flow(name="foo")
async def does_fail5():
    foos = foo.map(nums)
    return bar.map(foos)

Adding return seems to do the trick; again, I'm not a big fan of this because I have to track all my submitted/mapped tasks and make sure I return this. This could potentially lead to errors as the flow grows in task size or as it's modified with new tasks.

In those last few flow examples, I understand that I could call result() on foos and this might? end up failing the flow, however in that scenario none of my bar tasks have been executed, even if some of them could have been. I.e, I expect the flow-run to look like:

not

---

Ok that was a lot! I really enjoy Prefect, but all of the above leaves me feeling less than confident when I author flows. I'm wondering if perhaps a remedy for this is more documentation around scenarios where a failed task will NOT fail a flow. Or a flow arg like fail_flow_if_any_task_fails 🤣

[mattiamatrix]

I don't know if it helps, but I am having similar issues #18373

I noticed that I had unexpected behaviours when a .map function is the last operation in the flow. And, I also don't do return statements at the end of the flow.

Now, what it looks like is doing the trick (fingers crossed 🤞) is to add a wait() at the end of the flow that waits for all the futures returned by the map function.

Something like this:

Prefect 2 code:

@flow()
def my_flow():
	foo.map(...)

to Prefect 3 code:

from prefect.futures import wait

@flow()
def my_flow():
	futures = foo.map(...)
	wait(futures)

[zzstoatzz]

just to add color to the comment below on rationale

this behavior is an explicit change from prefect 2.x, "this" meaning that you need to explicitly resolve terminal futures

see:

• https://docs.prefect.io/v3/concepts/task-runners#submit-tasks-to-a-task-runner
• https://youtu.be/k74tEYSK_t8?si=fTFh6epSYuvAQq8g&t=218

[cicdw]

Appreciate the feedback, I totally get it. We made these design choices because Prefect over time has leaned more and more into a workflows-as-code philosophy where your Python code defines exactly what happens, including how failures propagate. When you use .submit(), you're explicitly choosing asynchronous execution and taking responsibility for handling the futures - this gives you fine-grained control over error handling, retry logic, and partial failure scenarios that many real-world workflows require.

All distributed workflow engines require this explicit resolution of futures because implicit handling creates significant problems, particularly with memory management in long-running workflows with thousands of tasks. Integrating with these systems and layering in this implicit future resolution caused a lot of hidden complexity.