Thanks for the clear write-up — this lines up with how Prefect 3 + Dask behaves today. What you’re seeing is largely a characteristic of Dask’s scheduler fairness rather than a Prefect-specific bug: - Prefect’s DaskTaskRunner submits task calls to the Dask scheduler. - Prefect does not currently expose Dask’s per-task priority/annotations through

task.submit()

or

@task(...)

. - When a task fails and is retried by Prefect, it is re-submitted to Dask and can contend with pending mapped items. Within the same priority, Dask’s FIFO-style fairness can let “recently re-submitted” work grab slots, which can look like starvation of pending items. Practical ways to mitigate the starvation 1) Add retry backoff and jitter so retries don’t immediately re-enter the queue - This gives pending mapped work time to use available slots.

from prefect import task

@task(
    retries=3,
    retry_delay_seconds=[10, 30, 60],  # backoff
    retry_jitter_factor=0.5            # optional jitter
)
def risky_task(x):
    ...

2) Use Prefect’s concurrency primitives to keep “risky” tasks from monopolizing slots - This gates concurrent runs of that task across the flow run(s), letting other mapped items proceed. - Async version:

from prefect.concurrency.asyncio import concurrency
from prefect import task

@task
async def risky_task(x):
    async with concurrency("risky-step", capacity=2):
        return await do_work(x)

- Sync version:

from prefect.concurrency.sync import concurrency
from prefect import task

@task
def risky_task(x):
    with concurrency("risky-step", capacity=2):
        return do_work(x)

3) Bound in-flight work with chunked mapping - Keep the active batch manageable; wait for each batch to settle (including retries) before moving on.

from prefect import flow, task

def batched(iterable, size):
    for i in range(0, len(iterable), size):
        yield iterable[i:i+size]

@task
def risky_task(x): ...

@flow
def my_flow(items):
    for batch in batched(items, 50):
        futs = risky_task.map(batch)
        # force completion of the current batch before queuing more
        [f.result() for f in futs]

4) Scale/shape the Dask cluster to improve available parallelism - More workers or different threads/processes per worker can reduce the visible impact of retries if you have headroom to scale. - You can also capture a Dask performance report to confirm where time is being spent:

from prefect_dask import DaskTaskRunner

runner = DaskTaskRunner(
    # your cluster settings...
    performance_report_path="dask-report.html"
)

Open the generated report to visualize scheduling and task queues. 5) Be selective about retries - Use

retry_condition_fn

to only retry on transient errors, and fail fast on others. This reduces “churn” in the scheduler from repeated retries. Notes on “why” and deeper options - Dask priorities and annotations exist (see Client.submit and Dask annotations), but Prefect’s task API does not expose them per-task today. As a result, you can’t easily give new, pending work a higher priority than retries without customizing the runner. - Advanced (unsupported) option: if you control your runtime, you could subclass the Dask task runner/client to derive a Dask

priority

from task metadata (e.g., tags) and pass it at submit time, or adjust Dask’s FIFO behavior. This requires custom code and careful testing. - Background on Dask scheduling/fairness: Scheduling policies If you can share: - How many items you’re mapping over - Your Dask cluster shape (n_workers, threads_per_worker, processes) - The task’s

retries

and

retry_delay_seconds

settings - A small repro (even synthetic)

…I’m happy to sanity-check and help tune concrete values (batch size, concurrency capacity, retry delays) for your workload. If we can reproduce a clear starvation pattern that persists even with backoff and bounded concurrency, I can help open a GitHub issue with a minimal example.

Got it — for your mapped check_files tasks, you want any slots freed by failures (headed into retry) to be used by other pending items instead of retries immediately grabbing them. The simplest, effective fixes: 1) Add non-zero retry delays (with jitter) - This prevents immediate re-queueing of failed items so pending items use the open slots.

from prefect import task

@task(
    retries=3,
    retry_delay_seconds=[10, 30, 60],  # backoff
    retry_jitter_factor=0.5            # optional jitter to avoid thundering herd
)
def check_file(path):
    return actually_check(path)

2) Gate “retries” with a small concurrency cap so they can’t monopolize workers - Use Prefect’s concurrency context and the task’s attempt number to allow first attempts to use full parallelism while keeping retries limited. - Async variant:

from prefect import task
from prefect.concurrency.asyncio import concurrency
from prefect.runtime import task_run

@task(retries=3, retry_delay_seconds=[10, 30, 60], retry_jitter_factor=0.5)
async def check_file(path):
    # First attempts get a larger pool; retries get a smaller pool
    is_retry = (task_run.run_count > 0)
    limit_name = "check-files-retries" if is_retry else "check-files-first"
    capacity = 4 if is_retry else 16  # tune to your cluster size

    async with concurrency(limit_name, capacity=capacity):
        return await actually_check_async(path)

- Sync variant:

from prefect import task
from prefect.concurrency.sync import concurrency
from prefect.runtime import task_run

@task(retries=3, retry_delay_seconds=[10, 30, 60], retry_jitter_factor=0.5)
def check_file(path):
    is_retry = (task_run.run_count > 0)
    limit_name = "check-files-retries" if is_retry else "check-files-first"
    capacity = 4 if is_retry else 16

    with concurrency(limit_name, capacity=capacity):
        return actually_check(path)

This pattern ensures, for example, that if 8 tasks fail and are going to retry, the remaining slots (and new slots as they free up) are primarily used by first-attempt items, while only a small number of retries run in parallel. 3) If you still see contention, add light batching to smooth pressure - Keep batches roughly equal to your effective parallelism:

def batched(xs, n):
    for i in range(0, len(xs), n):
        yield xs[i:i+n]

@flow
def main(file_paths):
    for batch in batched(file_paths, 16):  # tune to cluster size
        futs = check_file.map(batch)
        [f.result() for f in futs]         # wait per-batch

Why this works - In Prefect 3 with DaskTaskRunner, retries are resubmitted to Dask. Without delay/backoff, they re-enter the queue immediately and can contend with (or starve) pending mapped work. Prefect does not expose Dask task priorities directly through `@task`/`.map`, so the above patterns create scheduling “fairness” by time (backoff) and by limiting concurrency for retries. If you share: - Current `retries`/`retry_delay_seconds` for check_files - Dask cluster shape (n_workers, threads_per_worker, processes) - Typical batch size and failure rate …I can help you pick concrete numbers (retry backoff, retry capacity, batch size) so that when 8 fail, the other 8 start right away on your cluster. Useful refs: - Retries: Task retries - Concurrency: Concurrency limits - Dask scheduling/fairness background: Dask scheduling policies

Thanks for the snippet — this explains why the “non‑zero delay” didn’t help. Your

retry_jitter_factor=5

is the culprit. Prefect applies jitter using a clamped exponential (Poisson) distribution around your base delay. With such a large jitter factor (5), the lower bound of the clamp becomes very small. That means some retries are sampled to run almost immediately (near 0s), so they re-enter the queue and contend with pending items — exactly what you’re seeing. What to change - Use a modest jitter (e.g., 0.2–0.5) or remove jitter entirely if your goal is fairness over randomness. - Keep your backoff list as-is. Example

@task(
    task_run_name="check_files_{params[0].date}_{params[1]}_{params[4]}",
    retries=4,
    timeout_seconds=30,
    retry_delay_seconds=[10, 30, 60],   # backoff
    retry_jitter_factor=0.3             # small jitter; or set to None/0 to disable
)
def check_file(path):
    ...

To really guarantee pending items start when others fail, combine that with a small concurrency gate for retries so they can’t monopolize slots:

from prefect import task
from prefect.concurrency.sync import concurrency
from prefect.runtime import task_run

@task(
    retries=4,
    retry_delay_seconds=[10, 30, 60],
    retry_jitter_factor=0.3   # or 0/None
)
def check_file(path):
    # First attempts get more slots; retries get fewer
    is_retry = task_run.run_count > 0
    limit_name = "check-files-retries" if is_retry else "check-files-first"
    capacity = 4 if is_retry else 16  # tune to your cluster size

    with concurrency(limit_name, capacity=capacity):
        return actually_check(path)

This ensures that when 8 fail and head to retry, the “first-attempt” pool still has capacity for the other 8 pending items to start immediately, while only a few retries run at a time. If you want to confirm behavior: - Check Prefect logs/UI for each failed task — you should see “Retrying in ~X seconds at …”. If X is very small, that’s the jitter effect. - After reducing jitter, you should see scheduled times in the future and Dask workers picking up pending items right away. Docs for reference: - Task retries: Task retries - Concurrency limits: Concurrency

Short answer: with DaskTaskRunner, Prefect’s map submits all mapped child tasks immediately; there is no built-in “submit only N at a time.” If you’re seeing only 16 tasks start, that’s Dask running 16 at a time because of available workers/threads; the rest are already queued at the Dask scheduler and will start as workers free up. Details and how to verify - Prefect’s base TaskRunner.map() loops over the entire iterable and calls submit for each item. DaskTaskRunner delegates to that; there’s no batching. All items are sent to the Dask scheduler right away. - Dask then executes up to the capacity of your cluster (n_workers × threads_per_worker for threads; or processes). That’s why you observe “only 16 submit now” — the others are queued in Dask, not unsubmitted. How to confirm - Open the Dask dashboard (scheduler URL) and look at the Task Stream/Progress. You should see a large number of pending/queued tasks beyond the 16 running. - In Prefect UI, you’ll see many task runs in “Running” or “Pending/Submitted to infra” depending on timing; they correspond to the Dask futures already created. If you actually want to limit submissions (to bound graph size or control in-flight work) - Batch your mapping explicitly:

def batched(xs, n):
    for i in range(0, len(xs), n):
        yield xs[i:i+n]

@flow
def main(paths):
    for batch in batched(paths, 200):   # controls the number of in-flight items
        futs = check_file.map(batch)
        # Optionally wait per-batch to exert stronger backpressure
        [f.result() for f in futs]

- Or impose concurrency at the task level:

from prefect.concurrency.sync import concurrency
from prefect import task

@task
def check_file(path):
    with concurrency("check-files", capacity=16):  # match your intended in-flight cap
        return do_check(path)

If your concern is starvation on retries - Keep using backoff (with small or no jitter). - Add a small dedicated concurrency pool for retries vs first attempts (as shared earlier) so first attempts continue to fill free slots. If you share your Dask cluster config (n_workers, threads_per_worker, processes) and typical map size, I can suggest concrete batch sizes or concurrency capacities to get the behavior you want.

Great question. In Prefect 3.x there isn’t a “ConcurrentTaskRunner” class. You’re probably thinking of Prefect 2.x’s ConcurrentTaskRunner. In Prefect 3.x the equivalents are: - ThreadPoolTaskRunner (threads; good for I/O-bound work) - ProcessPoolTaskRunner (processes; good for CPU-bound work) - DaskTaskRunner (distributed via Dask; local or remote cluster) How they differ - ThreadPoolTaskRunner - Runs tasks in a local thread pool inside the flow process - Best for I/O-bound tasks (network, file I/O) - Simple, zero cluster overhead, predictable fairness with retry delays - Limited by the Python GIL for CPU-heavy tasks - Configure with max_workers to cap parallelism - ProcessPoolTaskRunner - Runs tasks in a local process pool (no GIL contention) - Better for CPU-bound tasks - Some serialization/startup overhead; higher memory use than threads - Also configured with max_workers - DaskTaskRunner - Submits tasks to a Dask distributed scheduler - Scales across processes and machines; dashboard, resources, worker targeting, autoscaling - More moving parts and scheduling nuances (e.g., the fairness behavior you’re seeing with retries) - Best for very large parallel maps or multi-node scale Is it fine to use “ConcurrentTaskRunner”? - In Prefect 3.x: use ThreadPoolTaskRunner (I/O-bound) or ProcessPoolTaskRunner (CPU-bound). Yes, it’s totally fine and often preferable if you don’t need Dask’s distributed features. - Given your goal (“use all 16 slots and let pending items run when others fail”), the thread or process pool runners tend to behave more predictably with retry delays — pending items will fill open slots while retries wait their scheduled delay. Quick examples - I/O-bound (most “check file” tasks fall here):

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

@task(retries=4, retry_delay_seconds=[10, 30, 60], retry_jitter_factor=0.3)
def check_file(path):
    return actually_check(path)

@flow(task_runner=ThreadPoolTaskRunner(max_workers=16))
def main(paths):
    return check_file.map(paths)

- CPU-bound:

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

@task(retries=4, retry_delay_seconds=[10, 30, 60], retry_jitter_factor=0.3)
def crunch(path):
    return heavy_compute(path)

@flow(task_runner=ProcessPoolTaskRunner(max_workers=16))
def main(paths):
    return crunch.map(paths)

When to prefer DaskTaskRunner - You need to scale beyond a single node - You want the Dask dashboard, worker resources, or adaptive scaling - You have very large task graphs that benefit from Dask’s scheduler features If your primary need is “run 16 in parallel and ensure pending items start when others fail,” switching to ThreadPoolTaskRunner (or ProcessPoolTaskRunner) is a good fit and should avoid the Dask retry/fairness quirk you observed.