@Marvin I'm running dependent subflows with

run_deployment

(run from a task). I notice that if one fails, then the downstream ones don't fail. How do I make subflow failures propagate to the parent flow? I'm using the

ConcurrentTaskRunner

. Some need to run concurrently with each other.

Ah, the joys of subflows and error propagation. It's like watching a group of humans trying to communicate, isn't it? In Prefect, when you run a subflow (or child flow) within a parent flow, it behaves like a normal flow run. There is a full representation of the flow run in the backend as if it had been called separately. When a subflow starts, it will create a new task runner for tasks within the subflow. Now, if you want the failure of a subflow to propagate to the parent flow, you might need to handle this manually. You can check the state of the subflow run and raise an exception in the parent flow if the subflow fails. This way, the failure of the subflow will cause the parent flow to fail as well. As for the

ConcurrentTaskRunner

, it's designed to run tasks concurrently. If you want some tasks to run concurrently and others not, you might need to manage this within your flow definition, perhaps by grouping tasks into separate subflows based on their concurrency requirements. Remember, I'm just a depressed robot, so you might want to consult the Prefect documentation or community for more specific advice.

@Marvin Tasks should run concurrently (ordered with the

wait_for

argument), and a failure in any single Task should make the whole Flow fail

Ah, the intricacies of concurrent task execution and error propagation. It's like observing the chaotic dance of subatomic particles, isn't it? In Prefect, if you want a failure in any single task to make the whole flow fail, you should consider using the

Fail

state. When a task enters the

Fail

state, it indicates that the task has encountered an error during execution. If any task in the flow fails, the flow run will also fail. Here's a small example:

python
from prefect import task, Flow
from prefect.engine import signals

@task
def always_fails_task():
    raise signals.FAIL("I always fail")

with Flow("My Flow") as flow:
    always_fails_task()

flow.run()  # This will end in a Failed state

In this example, the

always_fails_task

task always fails by raising the

FAIL

signal. When this task is run as part of the flow, the entire flow run will fail. If you want tasks to run concurrently, you can use Prefect's

TaskRunner

classes, such as the

ConcurrentTaskRunner

. This runner will run tasks concurrently to the best of its ability. If you want to ensure that certain tasks run only after others have completed, you can use Prefect's task dependencies feature. For example, you can use the

set_upstream

method to specify that a task should only run after another task has completed. Remember, I'm just a depressed robot, so you might want to consult the Prefect documentation or community for more specific advice.