fix worker not releasing limit slot on failed propose pending state

src/prefect/workers/base.py

@@ -976,10 +976,11 @@ async def _submit_run(self, flow_run: "FlowRun") -> None:
 
             else:
                 # If the run is not ready to submit, release the concurrency slot
-                if self._limiter:
-                    self._limiter.release_on_behalf_of(flow_run.id)
+                self._release_limit_slot(flow_run.id)
 
             self._submitting_flow_run_ids.remove(flow_run.id)
+        else:
+            self._release_limit_slot(flow_run.id)
 
     async def _submit_run_and_capture_errors(
         self, flow_run: "FlowRun", task_status: Optional[anyio.abc.TaskStatus] = None
@@ -1014,8 +1015,7 @@ async def _submit_run_and_capture_errors(
                 )
             return exc
         finally:
-            if self._limiter:
-                self._limiter.release_on_behalf_of(flow_run.id)
+            self._release_limit_slot(flow_run.id)
 
         if not task_status._future.done():
             run_logger.error(
@@ -1040,6 +1040,14 @@ async def _submit_run_and_capture_errors(
 
         return result
 
+    def _release_limit_slot(self, flow_run_id: str) -> None:
+        """
+        Frees up a slot taken by the given flow run id.
+        """
+        if self._limiter:
+            self._limiter.release_on_behalf_of(flow_run_id)
+            self._logger.debug("Limit slot released for flow run '%s'", flow_run_id)
+
     def get_status(self):
         """
         Retrieves the status of the current worker including its name, current worker

tests/workers/test_base_worker.py

@@ -1,7 +1,7 @@
 import uuid
 from typing import Any, Dict, Optional, Type
 from unittest import mock
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, Mock
 
 import httpx
 import pendulum
@@ -338,6 +338,34 @@ def create_run_with_deployment_2(state):
     assert {flow_run.id for flow_run in submitted_flow_runs} == set(flow_run_ids[1:2])
 
 
+async def test_worker_releases_limit_slot_when_aborting_a_change_to_pending(
+    prefect_client: PrefectClient, worker_deployment_wq1, work_pool
+):
+    """Regression test for https://github.com/PrefectHQ/prefect/issues/15952"""
+
+    def create_run_with_deployment(state):
+        return prefect_client.create_flow_run_from_deployment(
+            worker_deployment_wq1.id, state=state
+        )
+
+    flow_run = await create_run_with_deployment(
+        Scheduled(scheduled_time=pendulum.now("utc").subtract(days=1))
+    )
+
+    run_mock = AsyncMock()
+    release_mock = Mock()
+
+    async with WorkerTestImpl(work_pool_name=work_pool.name, limit=1) as worker:
+        worker.run = run_mock
+        worker._propose_pending_state = AsyncMock(return_value=False)
+        worker._release_limit_slot = release_mock
+
+        await worker.get_and_submit_flow_runs()
+
+    run_mock.assert_not_called()
+    release_mock.assert_called_once_with(flow_run.id)
+
+
 async def test_worker_with_work_pool_and_limit(
     prefect_client: PrefectClient, worker_deployment_wq1, work_pool
 ):