Control deployment concurrency limits with an orchestration policy

[abrookins]

Describe the current behavior

In #15085 and #15022, we made workers aware of deployment concurrency limits. This placed our ability to release a concurrency slot closest to our knowledge of when and why a flow run succeeded or failed, so we could reliably release a concurrency slot.

However, this has the tradeoff that for users to use deployment concurrency limits, they need to upgrade their workers to use the latest version of Prefect. We want to support users who do not or cannot upgrade their workers.

Describe the proposed behavior

The best approach to accomplish the goal of enabling older clients to use deployment concurrency limits is to move concurrency handling into new SecureFlowRunConcurrencySlots and ReleaseFlowRunConcurrencySlots rules in the core orchestration policy for flows. If we handle deployment concurrency as an orchestration rule, workers' attempts to propose states like Pending or Failed would acquire (or release) concurrency slots server-side.

For workers to propose a new state for a flow run, they need to have the same knowledge they would need to acquire or release a concurrency slot. Here are some cases that will still work the same with an orchestration policy instead of worker logic for deployment concurrency limits:

1. A worker needs to acquire a concurrency slot before spinning up expensive infrastructure for a flow run. This will happen transparently to the worker when it proposes Pending and receives a rejection because a slot isn't available. The orchestration policy will reschedule the run to try again.
2. A flow.serve() worker runs a flow with a deployment concurrency limit. No special handling is needed here because, just like other workers, when the worker proposes a Pending state, orchestration logic will acquire a slot.
3. A worker knows that the infrastructure running a flow crashed. It can propose a Crashed state for the flow, which will transparently release a concurrency slot.

Note: Prefect 2's task run concurrency hooked into the Running state. However, in this case, we want to prevent setting up flow run infrastructure if we don't have a concurrency slot. So this proposal uses hooks into the Pending state to acquire a concurrency slot.

Example Use

No response

Additional context

No response