https://prefect.io
Join Slack
<@ULVA73B9P> I have a flow running in Prefect 3 where the data is processed in three parts: one part...
# ask-marvin
r
@Marvin I have a flow running in Prefect 3 where the data is processed in three parts: one part in SQL, one part in Python, and then another part in SQL. To distribute the load, the maximum range is one day. Normally, the flow is triggered every minute. I want only one instance of the flow to run at a time, and as soon as another flow is scheduled, I want it to be canceled until the other flow has finished running. The flow is allowed to take up to one hour to complete. I’ve now tried to implement this with
Copy code
concurrency_limit=ConcurrencyLimitConfig(
                    limit=1,
                    collision_strategy=ConcurrencyLimitStrategy.CANCEL_NEW
                )
But when i have a flow running, the first 2 or 3 flows after that will be cancelled (a good thing) and then we have a running flow again when the first flow is not completed yet. Besides this, i used 
timeout_seconds=3600
on the task decorator but it seems to be ignored because those task can run for hours without being cancelled. How can we tackle those two problems? Can you give me a detail solution for both of them?
m
✅ thought for 188.8 seconds
Yes — these are two separate problems: 1. why 
CANCEL_NEW
is letting later runs start while an earlier run is still active 2. why 
timeout_seconds=3600
on a task appears to be ignored Here’s the practical fix for both in Prefect 3.x. --- ## 1) Why 
concurrency_limit=1
+ 
CANCEL_NEW
is not fully preventing overlap What you configured is the right idea, but deployment concurrency in Prefect uses a lease under the hood. If the lease expires before the run finishes, Prefect can free that slot and allow a later scheduled run to start — even though the first run is still executing. That’s why you see: - first few scheduled runs get canceled - then another run starts - while the original run is still running ### Root cause Your flow likely runs longer than the default deployment concurrency lease / grace period. ### Fix Set a longer 
grace_period_seconds
 on the deployment concurrency config so the lease stays valid for the whole expected runtime. If your flow can run up to 1 hour, set the grace period a bit higher than that, for example 3900 or 4200 seconds. Example:
Copy code
python
from prefect import flow
from prefect.client.schemas.objects import (
    ConcurrencyLimitConfig,
    ConcurrencyLimitStrategy,
)

@flow
def my_flow():
    ...

if __name__ == "__main__":
    my_flow.deploy(
        name="my-deployment",
        work_pool_name="my-pool",
        concurrency_limit=ConcurrencyLimitConfig(
            limit=1,
            collision_strategy=ConcurrencyLimitStrategy.CANCEL_NEW,
            grace_period_seconds=4200,
        ),
    )
### Why this helps With 
limit=1
and `CANCEL_NEW`: - if one run is active, new scheduled runs should be canceled - but only as long as the active run still holds its lease - 
grace_period_seconds
prevents the lease from expiring too early ### Important checks Also make sure: - your Prefect client/worker version is reasonably up to date - especially if you're on older 3.x versions, lease renewal behavior may be worse - all runs are from the same deployment whose concurrency limit you configured --- ## Recommended config for your use case Since you schedule every minute and runs may last up to an hour:
Copy code
python
concurrency_limit=ConcurrencyLimitConfig(
    limit=1,
    collision_strategy=ConcurrencyLimitStrategy.CANCEL_NEW,
    grace_period_seconds=4200,
)
I’d give it some buffer over 3600 seconds. --- ## 2) Why 
timeout_seconds=3600
on a task seems ignored This is also expected in some cases. 
timeout_seconds
on 
@task
is not a guaranteed OS-level kill. It depends on how the task is executed. ### Most common reason If your task is: - synchronous - running via the default thread-based task runner - and doing blocking work like: - SQL query - 
requests
- 
time.sleep
- blocking DB driver call then Prefect often cannot interrupt it immediately. So the task may continue running for hours even though 
timeout_seconds=3600
is set. --- # Best way to enforce timeouts ## A. Put a timeout on the flow too If the whole run must not exceed 1 hour, put 
timeout_seconds
on the flow itself:
Copy code
python
from prefect import flow, task

@task
def sql_part_1():
    ...

@task
def python_part():
    ...

@task
def sql_part_2():
    ...

@flow(timeout_seconds=3600)
def my_flow():
    sql_part_1()
    python_part()
    sql_part_2()
### But note Flow timeout is still not a perfect “kill switch” for a blocking task already in progress. It’s useful, but not sufficient by itself. --- ## B. Enforce timeout at the database/client level For long SQL steps, this is the most important fix. If your SQL query can run forever, Prefect may not be able to stop it from Python. You should set a DB-native statement/query timeout. ### Example for Postgres ```python from prefect import task import sqlalchemy as sa @task def sql_part(): engine = sa.create_engine("postgresql+psycopg2://...") with engine.begin() as conn:
conn.execute(sa.text("SET statement_timeout = 3600000")) # 1 hour in ms conn.execute(sa.text(""" SELECT ... """))
Copy code
If you want tighter limits per SQL phase, use smaller values, e.g. 20 minutes.

### Why this matters
Database-native timeout is much more reliable than Python-side cancellation for blocking SQL.

---

## C. If tasks are blocking sync code, consider `ProcessPoolTaskRunner`
If you are using `.submit()` and long-running synchronous tasks, a process-based task runner is often better than the default thread pool.

Example:

```python
from prefect import flow, task
from prefect.task_runners import ProcessPoolTaskRunner

@task(timeout_seconds=3600)
def blocking_sql_task():
    ...

@flow(task_runner=ProcessPoolTaskRunner(), timeout_seconds=3600)
def my_flow():
    blocking_sql_task.submit()
### Why A blocking sync task in a separate process is easier to interrupt than one stuck in a worker thread. ### Caveat Even then, I still strongly recommend DB-native query timeouts. That is the real protection for SQL. --- ## D. If possible, use async drivers for async tasks If your DB access is async and your code properly awaits, Prefect timeouts are more reliable. But if your codebase is currently sync SQL + sync Python, DB-native timeouts are usually the easiest and most robust path. --- # Practical solution for your case ## Problem 1: only one flow instance at a time Use deployment concurrency with a longer grace period:
Copy code
python
concurrency_limit=ConcurrencyLimitConfig(
    limit=1,
    collision_strategy=ConcurrencyLimitStrategy.CANCEL_NEW,
    grace_period_seconds=4200,
)
## Problem 2: prevent hour-long hangs Use all three: 1. 
@flow(timeout_seconds=3600)
on the whole flow 2. DB-native timeouts on each SQL section 3. optionally 
ProcessPoolTaskRunner
if your tasks are sync/blocking and submitted --- # Example combined pattern
Copy code
python
from prefect import flow, task
from prefect.task_runners import ProcessPoolTaskRunner
import sqlalchemy as sa

@task(timeout_seconds=1800)
def sql_part_1():
    engine = sa.create_engine("postgresql+psycopg2://...")
    with engine.begin() as conn:
        conn.execute(sa.text("SET statement_timeout = 1800000"))
        conn.execute(sa.text("SELECT ..."))

@task(timeout_seconds=1200)
def python_part():
    # your python processing
    ...

@task(timeout_seconds=1800)
def sql_part_2():
    engine = sa.create_engine("postgresql+psycopg2://...")
    with engine.begin() as conn:
        conn.execute(sa.text("SET statement_timeout = 1800000"))
        conn.execute(sa.text("SELECT ..."))

@flow(
    timeout_seconds=3600,
    task_runner=ProcessPoolTaskRunner()
)
def my_flow():
    sql_part_1()
    python_part()
    sql_part_2()

if __name__ == "__main__":
    my_flow.deploy(
        name="daily-window-flow",
        work_pool_name="my-pool",
        concurrency_limit={
            "limit": 1,
            "collision_strategy": "CANCEL_NEW",
            "grace_period_seconds": 4200,
        },
    )
--- # Extra note: if you truly want every new scheduled run to be dropped Your stated requirement is:
if another flow is scheduled, I want it to be canceled until the other flow has finished running
That is exactly what 
CANCEL_NEW
is for. So I would not switch to 
ENQUEUE
. The main thing missing is the longer 
grace_period_seconds
. --- # CLI reference If you are deploying through the CLI, use 
prefect deploy
in Prefect 3.x, not the old 2.x deployment build command: Prefect docs Deployments guide I also verified the CLI command exists:
Copy code
bash
prefect deploy --help
--- # My recommendation in one line For your case, set 
grace_period_seconds
> max flow runtime, add 
timeout_seconds
to the flow, and enforce SQL statement timeouts in the database driver/session. ---
If you want, I can help you with a concrete deployment snippet for your exact database stack (Postgres / SQL Server / Snowflake / BigQuery / etc.) and show the correct timeout configuration for that driver.
r
@Marvin how long is the normal grace_period_seconds?
m
✅ thought for 7.0 seconds
The normal/default 
grace_period_seconds
for deployment concurrency leases is 300 seconds — 5 minutes. So if you do not set it explicitly, Prefect uses about 5 minutes, which is why a flow that runs much longer can release its slot too early. For your case, where runs may take up to an hour, you should set it explicitly to something larger, for example:
Copy code
python
grace_period_seconds=4200
That gives you a 70-minute lease. If you want, I can also show you how to choose the right grace period based on your schedule frequency and worst-case runtime.
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