@Marvin is it expected that a flow with a timeout will wait for the actively running task to finish before terminating the flow? For example, in the below snippet the task will always complete before the flow is terminated and marked failed for exceeding the timeout.

@task
async def hanging_task():
    await asyncio.sleep(30)
    print("Finally finished.")


@flow(timeout_seconds=10)
async def impatient_flow():
    await hanging_task()
    # The above task completes before the flow will terminated due to timeout.
    # The below never prints as expected.
    print("Task completed without timeout.")

In Prefect 3.x, when a flow exceeds its specified timeout, a timeout exception is raised, and the flow is marked as failed. However, it seems that the flow timeout does not immediately interrupt or terminate any tasks that are actively running within the flow. Instead, the flow waits for the currently running task to complete before the timeout exception is raised and the flow is marked as failed. This behavior aligns with your observation that the

hanging_task

completes before the flow is terminated due to the timeout. The flow timeout acts at the level of the entire flow execution, not on individual tasks. For more details, you can refer to the Prefect documentation on cancellation and timeouts. If you want to enforce a timeout on individual tasks, you might need to handle it within the task itself or use task-specific timeout mechanisms.