https://prefect.io
Join Slack
<@ULVA73B9P> On Prefect 3.6.12, say I have a parent flow run which launches two subflows, all of the...
# ask-marvin
n
@Marvin On Prefect 3.6.12, say I have a parent flow run which launches two subflows, all of them on Kubernetes. these results in three pods running. If one of the subflows fails, the parent run also fails because it calls .result() on its flowrun object. What happens to the other subflow? Will it be cancelled? Or will it run to completion even if the parent flow run has failed!
m
✅ thought for 128.0 seconds
In your scenario, the other subflow will generally keep running to completion. It is not automatically cancelled just because: - one sibling subflow failed, and - the parent flow failed after calling 
.result()
on that failed subflow. A few important details: - In Prefect 3.x, failure propagation and cancellation propagation are different things. - A parent flow becoming FAILED does not automatically cascade cancellation to sibling subflows. - Automatic cascade behavior is tied to cancellation, not ordinary failure. So with: - parent flow in one pod - subflow A in another pod - subflow B in a third pod if subflow A fails and the parent does something like:
Copy code
python
a_state = a_future.result()
or otherwise awaits/reads that failed result and raises, then: - parent flow run -> FAILED - subflow A -> FAILED - subflow B -> keeps running unless something explicitly cancels it Because your subflows are on Kubernetes as separate flow runs / pods, they are independently orchestrated. The parent failing does not by itself tear down the other pod. If you want sibling subflows to stop when one fails, you need to do that explicitly — for example by catching the exception in the parent and issuing cancellation for the other flow run(s). Useful reference docs: - Flows - States - Run work concurrently If you want, I can show you a Prefect 3.6-compatible pattern for: 1. launching multiple subflows concurrently, and 2. cancelling the remaining ones if any one fails.
n
@Marvin show me the pattern
m
✅ thought for 188.2 seconds
Yep — here’s the safest Prefect 3.6.x pattern: - launch subflows concurrently - watch them as they finish - if one fails, explicitly request cancellation of the others Because you’re running subflows as separate flow runs / Kubernetes pods, you need to explicitly cancel those sibling flow runs. A good pattern is:
Copy code
python
import asyncio

from prefect import flow, get_run_logger
from prefect.client.orchestration import get_client
from prefect.client.schemas.objects import State, StateType


@flow
async def subflow_a():
    await asyncio.sleep(5)
    return "a-ok"


@flow
async def subflow_b():
    await asyncio.sleep(30)
    return "b-ok"


async def request_cancellation(flow_run_ids):
    async with get_client() as client:
        for run_id in flow_run_ids:
            if run_id is None:
                continue
            try:
                await client.set_flow_run_state(
                    flow_run_id=run_id,
                    state=State(type=StateType.CANCELLING),
                    force=True,
                )
            except Exception:
                # Best-effort cleanup
                pass


@flow
async def parent_flow():
    logger = get_run_logger()

    # Start subflows concurrently and ask Prefect to return State objects
    task_a = asyncio.create_task(subflow_a(return_state=True))
    task_b = asyncio.create_task(subflow_b(return_state=True))

    tasks = {task_a: "subflow_a", task_b: "subflow_b"}
    run_ids = {}

    try:
        for finished in asyncio.as_completed(tasks):
            state = await finished
            name = tasks[finished]

            child_run_id = state.state_details.child_flow_run_id
            run_ids[name] = child_run_id

            if state.is_failed() or state.is_crashed() or state.is_cancelled():
                logger.error(f"{name} ended in state {state.type}. Cancelling siblings.")

                sibling_run_ids = [
                    rid for subflow_name, rid in run_ids.items() if subflow_name != name
                ]

                # Also cancel any tasks we haven't seen finish yet, if their run ids
                # become available later via stored futures/state handling
                await request_cancellation(sibling_run_ids)

                raise RuntimeError(f"{name} failed with state {state.type}")

            <http://logger.info|logger.info>(f"{name} completed successfully")

        return "all subflows succeeded"

    except Exception:
        # Best-effort: cancel any child runs we already know about
        await request_cancellation(run_ids.values())
        raise
A few important notes: 1. Use 
return_state=True
on subflow calls That gives you a 
State
object instead of immediately raising on failure. Then you can inspect the state and decide what to do. 2. Cancel with 
StateType.CANCELLING
 That tells Prefect to begin cancellation of that child flow run. This is the right orchestration-level request. 3. Cancellation is best-effort For Kubernetes-backed runs, cancellation means Prefect asks the infrastructure/worker side to stop the run. It is not the same as instantly killing the pod at the exact line you call it. 4. Your subflows should handle cancellation cooperatively If a subflow is doing long blocking work, it may not stop immediately. Cancellation is usually picked up at orchestration/checkpoint boundaries. If you want a slightly more production-ready version, I’d actually recommend this structure instead: ```python import asyncio from prefect import flow, get_run_logger from prefect.client.orchestration import get_client from prefect.client.schemas.objects import State, StateType @flow async def worker(name: str, delay: int, fail: bool = False): await asyncio.sleep(delay) if fail: raise ValueError(f"{name} failed") return f"{name} succeeded" async def cancel_flow_runs(run_ids): async with get_client() as client: for run_id in run_ids: if not run_id:
continue try: await client.set_flow_run_state( flow_run_id=run_id, state=State(type=StateType.CANCELLING), force=True, ) except Exception: pass @flow async def orchestrator(): logger = get_run_logger() coros = [ worker("one", 10, fail=True, return_state=True), worker("two", 60, fail=False, return_state=True), ] pending = [asyncio.create_task(coro) for coro in coros] known_run_ids = [] try: for fut in asyncio.as_completed(pending): state = await fut child_run_id = state.state_details.child_flow_run_id if child_run_id: known_run_ids.append(child_run_id) if not state.is_completed(): logger.error( f"Detected failed child run {child_run_id} in state {state.type}; cancelling siblings." ) await cancel_flow_runs(known_run_ids) raise RuntimeError(f"Child flow failed: {state.type}") logger.info("All child flows completed successfully") except Exception: await cancel_flow_runs(known_run_ids) raise ``` One caveat with the above approach: - you only know a child run ID after you receive its returned state - if you want to cancel siblings immediately after submission, it’s better to work with submitted flow-run futures that expose 
flow_run_id
If you want, I can show you the even better version that uses subflow futures + 
flow_run_id
directly, which is usually the cleanest approach for your Kubernetes use case.
n
@Marvin show me the better version
m
✅ thought for 128.7 seconds
Here’s the better version for Prefect 3.6.x: use 
subflow.submit(...)
so you get a 
PrefectFlowRunFuture
immediately, including its 
flow_run_id
, then cancel siblings on first failure. This is nicer because you do not need to wait for a returned 
State
just to learn the child run ID.
Copy code
python
from prefect import flow, get_run_logger
from prefect.client.orchestration import get_client
from prefect.futures import as_completed
from prefect.states import Cancelled


@flow
def subflow_ok(name: str, delay_seconds: int = 30):
    import time

    time.sleep(delay_seconds)
    return f"{name} completed"


@flow
def subflow_fail(name: str, delay_seconds: int = 10):
    import time

    time.sleep(delay_seconds)
    raise RuntimeError(f"{name} failed")


def cancel_remaining(futures, exclude_flow_run_id=None):
    with get_client(sync_client=True) as client:
        for future in futures:
            if future.flow_run_id == exclude_flow_run_id:
                continue

            try:
                state = future.state
                if state is not None and state.is_final():
                    continue
            except Exception:
                # If we can't read state, still try cancellation
                pass

            try:
                client.set_flow_run_state(
                    flow_run_id=future.flow_run_id,
                    state=Cancelled(),
                    force=True,
                )
            except Exception:
                # Best-effort cleanup
                pass


@flow
def parent_flow():
    logger = get_run_logger()

    futures = [
        subflow_ok.submit("subflow-a", delay_seconds=60),
        subflow_fail.submit("subflow-b", delay_seconds=15),
        subflow_ok.submit("subflow-c", delay_seconds=120),
    ]

    for future in futures:
        <http://logger.info|logger.info>(f"Submitted child flow run: {future.flow_run_id}")

    results = []

    try:
        for completed_future in as_completed(futures):
            try:
                result = completed_future.result(raise_on_failure=True)
                results.append(result)
                <http://logger.info|logger.info>(
                    f"Child flow run {completed_future.flow_run_id} completed successfully"
                )
            except Exception as exc:
                logger.error(
                    f"Child flow run {completed_future.flow_run_id} failed: {exc}. "
                    "Cancelling sibling subflows."
                )

                cancel_remaining(
                    futures,
                    exclude_flow_run_id=completed_future.flow_run_id,
                )
                raise

        return results

    except Exception:
        # Optional extra cleanup pass
        cancel_remaining(futures)
        raise
Why this is the better pattern: - 
subflow.submit(...)
returns a 
PrefectFlowRunFuture
- 
future.flow_run_id
is available right away - 
as_completed(futures)
lets you react as soon as the first child finishes/fails - on failure, you can immediately call 
set_flow_run_state(...)
on all siblings A few practical notes: 1. Use 
Cancelled()
/ cancellation request as best-effort orchestration cleanup This tells Prefect to cancel the child flow runs. For Kubernetes-backed runs, that means the worker/infrastructure will be asked to stop them. It may not be instantaneous. 2. Don’t expect the pod to vanish the instant one sibling fails There can be a short delay between: - setting the child flow run to cancelled - the worker observing that state - the infrastructure/pod actually stopping 3. 
force=True
 I included 
force=True
because in failure cleanup paths it’s often what you want. If you prefer respecting orchestration rules more strictly, you can try 
force=False
. If your parent flow is async, the same idea still applies, but you’d typically use the async result methods. Example: ```python from prefect import flow, get_run_logger
from prefect.client.orchestration import get_client from prefect.futures import as_completed from prefect.states import Cancelled @flow async def async_subflow(name: str, fail: bool = False): import asyncio await asyncio.sleep(10) if fail: raise RuntimeError(f"{name} failed") return f"{name} completed" async def cancel_remaining_async(futures, exclude_flow_run_id=None): async with get_client() as client: for future in futures: if future.flow_run_id == exclude_flow_run_id: continue try: await client.set_flow_run_state( flow_run_id=future.flow_run_id, state=Cancelled(), force=True, ) except Exception: pass @flow async def async_parent_flow(): logger = get_run_logger() futures = [ async_subflow.submit("one"), async_subflow.submit("two", fail=True), async_subflow.submit("three"), ] try: for completed_future in as_completed(futures): try: result = await completed_future.aresult(raise_on_failure=True) logger.info(f"Success: {result}") except Exception as exc: logger.error(f"Failure from {completed_future.flow_run_id}: {exc}") await cancel_remaining_async( futures, exclude_flow_run_id=completed_future.flow_run_id, ) raise except Exception: await cancel_remaining_async(futures) raise ``` Docs: - Flows - States - Run work concurrently If you want, I can also show a version that is Kubernetes-worker specific, including how to make cancellation more responsive in long-running child flows.
Open in Slack
PreviousNext
Powered by

Prefect Community

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.

Powered by