Hi Prefect Team -- I have a question about the beh...
# ask-community
s
Hi Prefect Team -- I have a question about the behavior of subflows vs. flows. Sometimes I need to "retry" a deployed flow run which contains subflows. When I retry the outer flow, new subflow runs are created. This causes any completed tasks within the subflow to re-run. Ideally, I would like the retried flow to skip all completed tasks within the subflows. Is there any way to avoid this behavior? To reproduce this behavior, I have a deployed
main_flow
which calls a
subflow
which calls 2 sequential tasks. I cause the second task to fail. When I retry the main flow from the UI, it spawns a new
subflow
run instead of retrying the existing
subflow
. This causes
task1
in the
subflow
to re-execute, despite it's already succeeded state in the initial flow run.
Copy code
@flow
def main_flow():
    subflow.with_options(persist_result=True, result_storage=FLOW_STORAGE)()

@flow
def subflow():
    task1.with_options(persist_result=True, result_storage=FLOW_STORAGE)()
    task2.with_options(persist_result=True, result_storage=FLOW_STORAGE)()

@task
def task1():
    print("running task 1")

@task
def task2():
    print("running task 2")
    raise RuntimeError

# deploy main_flow() ....