fix worker not releasing limit slot on failed propose pending state

[jdjfisher]

Fixes a bug where workers aborting a flow run submission doesn't release the concurrency slot that was claimed. I've based this on what has been done in the runner as this issue doesn't seem to exist over there

prefect/src/prefect/runner/runner.py

Line 963 in 8758666

def _release_limit_slot(self, flow_run_id: str) -> None:

. I've had a quick look at putting together a test case for this but I might need a hand here. Cheers.

Closes #15952

[codspeed-hq]

CodSpeed Performance Report

Merging #16012 will not alter performance

_{Comparing jdjfisher:fix-worker-limiter-release-15952 (75d87f4) with main (377505b)}

Summary

✅ 3 untouched benchmarks

[desertaxle]

Thanks for the PR @jdjfisher!

I think there's one other place where your new release_on_behalf_of method could be used, which would be good for consistency.

For a test, I think you could write a test similar to this one:

prefect/tests/workers/test_base_worker.py

Line 341 in 86b80dc

async def test_worker_with_work_pool_and_limit(

and mock _propose_pending_state to return False so that you can ensure that the concurrency limiter slot is correctly released.

Let me know if you have any questions about either of those points!

[jdjfisher]

Thanks for the review @desertaxle, super helpful. I'm managed to put a case together based on your suggestions that should cover this fix.