Prefect Tasks crash with 429 errors before running

Hello all, I use prefect the last year or so and lately I have been experiencing some issues with my tasks, they keep crashing with a Client error ‘429 Too Many Requests’.

My workflow utilizes async to run multiple tasks similar to this example.

import asyncio
from prefect import task, flow

@task
async def print_values(values):
    for value in values:
        await asyncio.sleep(1) # yield
        print(value, end=" ")

@flow
async def async_flow():
    await print_values([1, 2])  # runs immediately
    coros = [print_values("abcd"), print_values("6789")]

    # asynchronously gather the tasks
    await asyncio.gather(*coros)

asyncio.run(async_flow())

Note, the code above from the documentation. In my case, each task may take several minutes to complete or even more (~10 min / task).

After monitoring my workflow, I found out that crashes occur whenever there are hundreds of Tasks to be run (>150 at least). I already have set concurrency limits to 10, so each Flow is only allowed to run a maximum of 10 tasks. Also, some Tasks start crashing before they are even awaited (i.e., flow is already running 10 tasks and they are in a ‘pending’ state)

Note, I use Prefect’s free Cloud version and I understand that rate limits exists. I just want to know whether Prefect somehow receives requests from all tasks that are waiting to be run (i.e., all coros and this is causing my flow to hit the limits).