Thanks for the details — that error means the flow/task lost its concurrency “slot” while Prefect was trying to renew the lease. When that happens, Prefect cancels execution to avoid over-allocating that resource. A few common causes and checks (assuming Prefect 3.x): What typically triggers this - A flow or task timeout fires and cancels the run; the lease renewal then fails and you see this message as a side-effect - Holding a tag-based concurrency lease for the entire flow (via flow tags) on a long run increases the chance of a transient network hiccup causing a renewal failure - An async task blocks the event loop for a long time (no awaits), preventing the background renewal coroutine from running - Transient network issues (e.g., ACI egress/NAT idle, corporate firewall) cause the periodic HTTP call to renew the lease to fail - The underlying concurrency limit is reset/deleted/changed while your run is active Quick checks 1) Look for an earlier timeout message in the run logs (e.g., “Flow run exceeded timeout of X seconds”). If present, raise/remove the timeout on the flow/task. 2) Verify whether the flow itself is tagged with a concurrency-limited tag. If yes, try moving the concurrency to only the long-running task(s) so the flow does not hold a lease for ~40 minutes. 3) Inspect your limits and active slots:

prefect concurrency-limit ls
prefect concurrency-limit inspect <your-tag>

Confirm that

active_slots

behaves as expected during the long task and that the limit isn’t being reset mid-run. Actionable mitigations - Only hold concurrency where it’s needed: - Avoid tagging the entire flow; tag just the task(s) that truly require the limit. - Example:

from prefect import flow, task
    from prefect.task_runners import ConcurrentTaskRunner

    @task(tags=["limited"])
    def long_task():
        # blocking work is fine if this is a sync @task (runs in a thread)
        ...

    @task
    def final_step():
        ...

    @flow(task_runner=ConcurrentTaskRunner())
    def pipeline():
        long_task()
        final_step()  # untagged; no concurrency lease here

- If you are using async tasks, make sure long operations yield periodically or run blocking work in a thread: - Wrap blocking code with

await anyio.to_thread.run_sync(func)

or add periodic

await anyio.sleep(0)

to avoid starving the renewal task. - Network hardening: - Ensure the ACI environment has reliable outbound access to your Prefect API (Cloud/Server). - If behind a firewall/proxy, increase idle/connection timeouts so periodic lease renewals aren’t dropped. - Optionally increase HTTP request timeouts in the container:

PREFECT_API_REQUEST_TIMEOUT=60

- Upgrade to the latest 3.x release — there have been ongoing improvements and bug fixes around leases and cancellation paths. - Do not reset or delete the concurrency limit tag while runs are active. Related references - Tag-based concurrency limits - Known reports for context: - Issue: lease renewal failed after a timeout/cancellation - Issue: infra lingering after failed lease renewal A couple clarifying questions to pinpoint your case: - Which Prefect version are you on? - Are you applying the concurrency tag at the flow level or only on the long-running task(s)? - Do you have

timeout_seconds

set on the flow or the long task? - Could you share the tag name you’ve limited and the output of:

prefect concurrency-limit inspect <your-tag>

- Any network proxy/firewall or VNET in front of your ACI that could impose idle timeouts? If you can share a minimal snippet of your flow definition (how tags/concurrency are applied) and a few lines of logs around the error, I can suggest a targeted fix.

@Marvin as my workload is mostly I/O bound and using asyncio I switched from ThreadpoolRunner to concurrent task runner, but I still have issues with the lease not being renewed for longer running flows. How can I fix that!

Thanks for the details — a couple of things can cause leases to lapse on long-running async flows. First, a quick clarification: - Are you on Prefect 3.x or 2.x? In 3.x there is no

ConcurrentTaskRunner

class — most users either keep things async/await-native or use

ThreadPoolTaskRunner

for sync/blocking work. Knowing your version and the exact log message (e.g., “deployment concurrency lease expired” vs “work pool lease expired”) will help pinpoint this. Why leases lapse with asyncio - In Prefect 3.x, lease renewal is tied to periodic heartbeats from the runner. If your flow/event loop is blocked (e.g., long sync calls,

time.sleep

, CPU-bound loops), heartbeats can’t fire and the lease expires. - By default, runner heartbeats are disabled in 3.x unless you turn them on. What to do 1) Enable runner heartbeats - Set an environment variable (minimum is 30s):

PREFECT_RUNNER_HEARTBEAT_FREQUENCY=30

- This makes the runner send periodic heartbeats so leases renew during long runs. - Docs: Settings reference (see RunnerSettings.heartbeat_frequency) 2) Make sure your async code yields to the event loop - Avoid

time.sleep()

or long, synchronous I/O in your async flows/tasks. - Prefer async libraries (e.g.,

httpx.AsyncClient

,

asyncpg

,

aiobotocore

) and

await

them. - If you must call sync/blocking functions from an async flow, run them in a thread:

import anyio

     async def call_blocking(fn, *args, **kwargs):
         return await anyio.to_thread.run_sync(lambda: fn(*args, **kwargs), cancellable=True)

- Even if you’re using

ThreadPoolTaskRunner

, a blocking synchronous flow function can still starve heartbeats if there are no await points. 3) If you’re on 3.x, prefer one of these patterns - Keep everything async/await-native (no special task runner needed), or - Use

ThreadPoolTaskRunner()

for sync I/O tasks to keep the event loop free:

from prefect import flow
     from prefect.task_runners import ThreadPoolTaskRunner

     @flow(task_runner=ThreadPoolTaskRunner())
     async def my_flow():
         # async code + offload any blocking calls to threads
         ...

- Concept doc: Task runners in Prefect 3.x 4) Don’t confuse worker heartbeat with flow-run heartbeat - Worker heartbeat/polling is separate and won’t renew your flow’s lease by itself. You still need the runner heartbeats above. - If you tweaked worker heartbeat intervals, that’s fine — but it does not replace runner heartbeats. 5) Upgrade to the latest 3.x - If you’re not on the latest 3.x, upgrade — there have been fixes and reliability improvements around heartbeats/engine behavior. Helpful diagnostics - Set logging to debug and look for “heartbeat” or “lease” log lines:

PREFECT_LOGGING_LEVEL=DEBUG

- Share the exact log message you see when the lease lapses; that will confirm if it’s a deployment concurrency lease or something else. - If you can, share your flow signature (sync vs async), the task runner you’re using, and an example of any long-running I/O call. Related docs - Heartbeats and zombie-flow detection: Detect and recover zombie flows - Settings (Runner heartbeat): Settings reference If you can paste: - Prefect version, - The exact error/log line about the lease, - A minimal example of your flow (sync/async) and how the long I/O is done, …I can give you a targeted fix.

Got it — thanks for the extra detail. Since you’re on Prefect 3.4.25 and most of your work is async, the pandas bits are the likely culprit. Synchronous pandas transformations inside an async flow will block the event loop, which prevents runner heartbeats from firing and can cause leases to expire. What to do - Enable runner heartbeats (off by default in 3.x):

PREFECT_RUNNER_HEARTBEAT_FREQUENCY=30

Link: Settings reference (RunnerSettings.heartbeat_frequency) - Offload pandas to a thread from your async flow so the event loop can keep sending heartbeats:

from prefect import flow
  from prefect.utilities.asyncutils import run_sync_in_worker_thread
  import pandas as pd

  def heavy_transform(df):
      # synchronous pandas code here
      return df.groupby("id").agg({"value": "sum"})

  @flow
  async def my_flow():
      # ... async I/O work
      df = await run_sync_in_worker_thread(heavy_transform, df)
      # ... continue async work

This is the recommended pattern; there isn’t an

as_thread

helper in Prefect 3.x — use

run_sync_in_worker_thread

(or

anyio.to_thread.run_sync

) for sync sections. - If your pandas is truly CPU-bound and heavy, consider a process pool instead of threads: - Option A: mark the transform as a

@task

and set

task_runner=ProcessPoolTaskRunner()

at the flow - Option B: use a process pool for just that step Threads are usually fine for pandas, but processes can help when the GIL is a bottleneck. - If you’re launching a large

asyncio.gather

, keep the async parts async (no

time.sleep

or blocking loops) and put the pandas bits behind

run_sync_in_worker_thread

. You can also yield periodically in long async loops:

import asyncio
  await asyncio.sleep(0)  # yields back to the event loop

Useful refs - Task runners overview: Task runners in Prefect 3.x - Zombie/heartbeat details: Detect and recover zombie flows If you share a small snippet showing how you call the pandas transform inside the async flow, I can suggest the exact offload placement. Also confirm you’ve set

PREFECT_RUNNER_HEARTBEAT_FREQUENCY=30

— that, plus offloading pandas, should keep your leases renewed.

Thanks — that detail helps. What you’re describing sounds like a global concurrency lease that expires during a long task, then the engine tries to release/renew right at the task boundary and hits an error when moving to the next task. A few ways to make this robust: - Hold the concurrency slot for the entire flow (avoid per-task acquire/release) - If your concurrency tag is on the tasks, each task acquires/releases a lease. That’s exactly where you’ll see boundary-time failures. - Move the tag to the flow so the flow holds a single slot the whole run:

from prefect import flow

    @flow(tags={"my-limit"})
    async def my_flow():
        # run all tasks; slot is held by the flow for the entire run
        ...

- This removes the rapid release/reacquire at task transitions entirely. - Docs: Global concurrency limits - Keep pandas off the event loop - You’re already mostly async; the pandas step can still block at the end of the task (result serialization, large in-memory operations). - Offload to a thread or process so heartbeats and concurrency lease renewals keep happening:

from prefect.utilities.asyncutils import run_sync_in_worker_thread

    def heavy_transform(df):
        # pandas work here
        return df  # transformed

    async def run_transform(df):
        return await run_sync_in_worker_thread(heavy_transform, df)

- For very CPU-heavy transforms, consider a process pool (e.g., make it a @task and use ProcessPoolTaskRunner for that task). - If you must keep the tag on tasks - Increase the decay on the limit so “denied slots” recover between tasks faster: - Edit your global concurrency limit and raise

slot_decay_per_second

so it decays quickly between task boundaries. This helps when a long-running task kept the limit saturated for a while. - Add a tiny yield between tasks so the event loop runs the renewal/cleanup promptly:

import asyncio
    await asyncio.sleep(0)

- Keep runner heartbeats enabled - You’ve already set this; keep it on:

PREFECT_RUNNER_HEARTBEAT_FREQUENCY=30

- Docs: Settings reference (RunnerSettings.heartbeat_frequency) - Check for heavy boundary work - Large DataFrame return values, big log bursts, or artifact uploads at the end of a task can stall the loop right when leases are being renewed/released. If that’s the case: - Persist large results externally (e.g., object store) and return small references. - Reduce log volume at task completion. If you can share: - The exact error/log line you see (“renew_concurrency_lease failed…”, “decrement_with_lease failed…”, or “unable to acquire slot…”), - Whether the concurrency tag is on the flow, the tasks, or both, - The name of the limit/tag and its current

slot_decay_per_second

, …I can give you a precise fix. But in practice, moving the tag to the flow (so you hold a single lease across the whole run) plus offloading pandas to a thread has been the most reliable way to eliminate these boundary-time lease issues.