Good Morning In Orion Prefect 2 0a4 I m starting to test out

Question

Good Morning In Orion Prefect 2 0a4 I m starting to test out subflows I m noticing that when using a subflow as a future added to a task s `wait for` argument Orion seems to ignore the wait for and processes the tasks even when the subflow fails

Kelby · Answer

```import sys from prefect import task flow sys tracebacklimit = 0 task def should not run print THIS SHOULD NOT APPEAR task def fail raise Exception Fail flow def subflow fail flow def my flow ss = fail should not run wait for=ss flow def my flow using subflow ss = subflow should not run wait for=ss if name == main print 50 print NO SUBFLOW print 50 my flow print 50 print WITH SUBFLOW print 50 my flow using subflow ```

Kelby · Answer

``` NO SUBFLOW 09 39 32 463 | Beginning flow run glorious teal for flow my flow 09 39 32 463 | Starting executor `SequentialExecutor` 09 39 32 518 | Submitting task run fail 6aa020f4 0 to executor 09 39 32 537 | Task run fail 6aa020f4 0 encountered exception Exception Fail 09 39 32 559 | Task run fail 6aa020f4 0 finished in state Failed message= Task run encountered an exception type=FAILED 09 39 32 579 | Submitting task run should not run 9f323c93 0 to executor 09 39 32 593 | Task run should not run 9f323c93 0 finished in state NotReady message= Upstream task run 0680e64f 2da8 4eaf 9c07 a0c2112dc753 did not reach a COMPLETED state type=PENDING 09 39 32 594 | Shutting down executor `SequentialExecutor` 09 39 32 608 | Flow run glorious teal finished in state Failed message= 1 2 states failed type=FAILED WITH SUBFLOW 09 39 32 638 | Beginning flow run glittering hound for flow my flow using subflow 09 39 32 638 | Starting executor `SequentialExecutor` 09 39 32 755 | Beginning subflow run tuscan numbat for flow subflow 09 39 32 755 | Starting executor `SequentialExecutor` 09 39 32 797 | Submitting task run fail 6aa020f4 1 to executor 09 39 32 814 | Task run fail 6aa020f4 1 encountered exception Exception Fail 09 39 32 831 | Task run fail 6aa020f4 1 finished in state Failed message= Task run encountered an exception type=FAILED 09 39 32 832 | Shutting down executor `SequentialExecutor` 09 39 32 855 | Subflow run tuscan numbat finished in state Failed message= 1 1 states failed type=FAILED 09 39 32 873 | Submitting task run should not run 9f323c93 1 to executor THIS SHOULD NOT APPEAR 09 39 32 908 | Task run should not run 9f323c93 1 finished in state Completed message=None type=COMPLETED 09 39 32 909 | Shutting down executor `SequentialExecutor` 09 39 32 922 | Flow run glittering hound finished in state Failed message= 1 2 states failed type=FAILED ```

Kelby · Answer

In the above code in `my flow` I have a task that fails so the `should not run` function should be skipped That is working as expected

Kelby · Answer

In `my flow using subflow` I introduce a subflow which has a task that fails Given that the subflow failed I expected that the `should not run` task would be skipped However it is still getting run Is this the expected behavior Thanks

Anna Geller · Answer

I will look at it in more detail but just to clarify in the flow below you want that `should not run` doesn t even start the execution if the child flow `ss` fails ``` flow def my flow using subflow ss = subflow should not run wait for=ss ```

Kelby · Answer

Correct Thanks

Anna Geller · Answer

< Kelby> I think this is how you can do it ``` flow def my flow using subflow ss = subflow ss result = ss result raise on failure=True should not run wait for=ss result ```

Anna Geller · Answer

or even this ``` flow def my flow using subflow ss = subflow ss result = ss result should not run wait for=ss result ``` I think the issue is that we need to explicitly wait until the child flow is finished and with result we do that

Kelby · Answer

Yes functionally that appears to work It does repeat the exception in the log though ``` WITH SUBFLOW 10 26 51 063 | Beginning flow run misty llama for flow my flow using subflow 10 26 51 063 | Starting executor `SequentialExecutor` 10 26 51 178 | Beginning subflow run rich turaco for flow subflow 10 26 51 178 | Starting executor `SequentialExecutor` 10 26 51 241 | Submitting task run fail 6aa020f4 1 to executor 10 26 51 262 | Task run fail 6aa020f4 1 encountered exception Exception Fail 10 26 51 285 | Task run fail 6aa020f4 1 finished in state Failed message= Task run encountered an exception type=FAILED 10 26 51 286 | Shutting down executor `SequentialExecutor` 10 26 51 320 | Subflow run rich turaco finished in state Failed message= 1 1 states failed type=FAILED 10 26 51 320 | Flow run misty llama encountered exception Exception Fail 10 26 51 321 | Shutting down executor `SequentialExecutor` 10 26 51 339 | Flow run misty llama finished in state Failed message= Flow run encountered an exception type=FAILED ```

Anna Geller · Answer

the exception is once logged for the task run and once for the flow run of the subflow therefore it s twice

Zanie · Answer

I think this is unintended

Zanie · Answer

You shouldn t have to throw an exception in the flow to prevent a downstream from running

Kelby · Answer

In my case it was the opposite If I have an exception in my subflow I don t want the later tasks to execute

Zanie · Answer

Yep I m saying unpacking the subflow exception with ` result ` to reraise it in the parent isn t something I want you to have to do

Kelby · Answer

Ahh Agreed

Zanie · Answer

We re only checking for failures in futures passed downstream not states

Zanie · Answer

We wanted people to be able to pass states downstream manually so they could work with them so we can t just stop your task from running if it receives failed states

Zanie · Answer

This is tricky with a subflow since it returns a state instead of a future

Anna Geller · Answer

< Zanie> how should it ideally be defined then like this ``` flow def flow with subflow ss = subflow should not run wait for=ss ``` if so I can open an issue