In Prefect 3.x, handling task timeouts and cancellations can be a bit tricky, but it is possible. Here's what I've gathered: 1. Task Cancellation: Prefect supports task cancellation, but there are conditions. The infrastructure running the task must support cancellation. If the infrastructure can't be found or has already been terminated, the task will be marked as canceled, but actual cancellation might not be enforced. More details can be found here. 2. Timeout Handling: The

TaskRunTimeoutError

is raised when a task exceeds its timeout. You can handle this exception within your task to perform any cleanup or logging before the task is marked as failed. The

task_engine.py

source code provides more insights into handling timeouts and task states. You can explore it here. 3. Concurrency with Async Tasks: Your idea of using

asyncio.wait

with

asyncio.FIRST_COMPLETED

is viable. You'd handle the cancellation logic within the async task that completes first. Prefect tasks can be defined as async functions, allowing you to integrate this logic directly into the task. For more detailed guidance, especially on configuring your infrastructure to support these features, you might need to refer to the infrastructure-specific documentation or community forums. If you have any specific issues or errors, feel free to share them, and I can assist further.