Kevin Hu
02/19/2026, 6:49 PMMarvin
02/19/2026, 6:49 PMMarvin
02/19/2026, 6:51 PMtimeout_seconds on the flow and/or tasks:
from prefect import flow, task
@task(timeout_seconds=120) # 2-minute cap per task attempt
def work():
...
@flow(timeout_seconds=600) # 10-minute cap for the whole flow run
def my_flow():
work()
- You can override at deploy time:
my_flow.with_options(timeout_seconds=900).deploy(
name="my-deployment",
work_pool_name="default"
)
- Note: timeouts mark the run as failed and attempt to cancel; truly preempting blocking sync code in threads isn’t always possible. Prefer async I/O or ProcessPoolTaskRunner for stronger preemption. Details: Task/flow timeout behavior
- Automations to cancel long runs (Cloud)
- Create an Automation that cancels any flow run that stays Running longer than X minutes:
from datetime import timedelta
from prefect.automations import Automation
from prefect.events.schemas.automations import EventTrigger
from prefect.events.actions import CancelFlowRun
Automation(
name="Cancel long-running flows",
trigger=EventTrigger(
after={"prefect.flow-run.Running"},
expect={"prefect.flow-run.Completed"},
posture="Proactive",
within=timedelta(minutes=30),
for_each={"prefect.resource.id"},
),
actions=[CancelFlowRun()],
).create()
- Metric-based triggers (e.g., “duration > N seconds”) are also available in Cloud.
- Docs: Create automations, Event triggers, Cancel workflows
If you can share whether you’re on Prefect Cloud or OSS, and whether you want a strict hard-stop at N minutes or just a cancellation after N minutes, I can suggest the best-fit pattern and example for your setup.