Hey all! Is there a way inside a flow using ConcurrentTaskRunner to specify "at most X tasks at once please". We've had a lot of issues using tag based concurrency (mostly slots not being released if a task crashes)

Hi Sam! The first thing that comes to mind as an alternative would be applying Global concurrency limits to your tasks

Hmm, its tough because we really only want to limit it for this single flow so we dont get a slap on the wrist re s3 rate limiting. An ideal API if this is ever considered would be something like:

@flow(task_runner=ConcurrentTaskRunner(maximum_concurrency=10))
def my_flow:
   for blah in blahs:
      my_task.submit(blah)