You’re seeing that behavior because your parent flow is async, but you’re calling two synchronous flows in the middle. Those sync calls block the event loop, so the background task won’t actually run until control returns to the event loop. A few good patterns in Prefect 3.x: Option A: Keep everything in-process and don’t block the event loop - Use

asyncio.create_task

(from Python) to start the async subflow - Offload the blocking sync flows to a worker thread so the event loop can keep running Example:

import asyncio
import anyio
from prefect import flow

@flow
async def contributed_datamart(target: str = "prod"):
    elastic_server = "prod"
    index_suffix = "dev"

    # Start async subflow immediately
    es_task = asyncio.create_task(
        elastic_sync(elastic_server=elastic_server, index_suffix=index_suffix)
    )

    # Run blocking sync flows without blocking the event loop
    await anyio.to_thread.run_sync(load_into_unifyd)
    await anyio.to_thread.run_sync(data_entry_tables_synch, target_database="canna_prod")

    # Wait for elastic sync to complete
    await es_task

Alternatively, you can run them all concurrently with gather:

import asyncio
import anyio
from prefect import flow

@flow
async def contributed_datamart(target: str = "prod"):
    await asyncio.gather(
        elastic_sync(elastic_server="prod", index_suffix="dev"),
        anyio.to_thread.run_sync(load_into_unifyd),
        anyio.to_thread.run_sync(data_entry_tables_synch, target_database="canna_prod"),
    )

Important notes: - Use

asyncio.create_task(...)

, not

create_task(...)

. - If

elastic_sync

is defined as a Prefect async flow (

@flow

with

async def

), calling it as shown will create a proper subflow run and it will be tracked in the UI. The

anyio.to_thread.run_sync(...)

calls keep the event loop responsive so your elastic sync can progress. Option B: Run the elastic subflow remotely (true background flow run) If you’re okay running

elastic_sync

on a work pool (separate worker), submit it and wait later:

from prefect import flow

@flow
def contributed_datamart(target: str = "prod"):
    es_future = elastic_sync.submit(elastic_server="prod", index_suffix="dev")
    load_into_unifyd()
    data_entry_tables_synch(target_database="canna_prod")
    es_future.result()  # wait at the end

Notes: -

.submit()

on flows submits to a work pool (remote execution) and returns a future. This won’t block your parent flow while the worker runs the subflow. - You’ll need a work pool + worker online for this to run. Which approach do you prefer? In-process concurrency (Option A) or running the elastic sync on a worker (Option B)? Also, can you confirm: - Is

elastic_sync

an async flow (

async def

)? - Are

load_into_unifyd

and

data_entry_tables_synch

definitely synchronous and CPU-bound (or just blocking I/O)?