I have a task which is annotated with a tag (based on the prefect quickstart code)

@task(tags=["process-customer"])
def process_customer(customer_id: str) -> str:
    <http://mylogger.info|mylogger.info>(f"Processing customer {customer_id}")
    time.sleep(4)
    return f"Processed {customer_id}"

[process_customer.submit(i) for i in range(300)]

I see that when a tag is preset, prefect will send concurrency increment/decrement lease requests with every task thread, even if I have no

tag:process-customer

concurrency value set. This has a pretty significant effect on runtime due to the rate limit on concurrency requests. For example on the free plan on prefect cloud (250/min limit) if I launch 300 tasks then it takes about 90 seconds to complete, and performs about 1400 concurrency requests, ~700 of which are retried 429s If I remove the tag then the 300 threads run in about 5 seconds (makes sense given the 4 second sleep + setup/teardown + 1 second logging granularity) Is there a way to make prefect not make a concurrency lease request if there is no concurrency with this tag name present? If not then we're probably going to need to abandon the use of tags and use something custom for our task annotation

(getting feedback from the team now but) let me know what you think about this!

thank you 🎉

thanks for the feedback! i'll pass it along please do let us know if you see any other places where the client is making extra calls