Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Prefect Community

Hi, does anyone know why this code leaves some subflows running?
```import asyncio
from typing import Any
from prefect import flow

@flow
async def subflow(should_fail: bool = False):
    if should_fail:
        raise ValueError("I failed!")
    return 42


@flow
async def create_sub_flows():
    parallel_subflows: list[Any] = []
    for i in range(10):
        parallel_subflows.append(subflow(i % 2 == 0))

    await asyncio.gather(*parallel_subflows)


if __name__ == '__main__':
    asyncio.run(create_sub_flows())```