Thanks for the clear summary — sounds like your flow finished its work but got stuck in “Running” because the server errored while setting the final state. I’ll help you unstick it now and then track down the root cause. First, a couple quick questions so I can tailor the fix: - Are you on Prefect Cloud or self-hosted Prefect Server? (Your note sounds self-hosted.) - Prefect version? (Assuming Prefect 3.x unless you say otherwise.) - What database is your server using (SQLite vs Postgres)? - Can you share the exact error from your server logs at the time it tried to set the state to Completed? Immediate ways to get the flow run unstuck - Manually mark it Completed via the API using the Prefect client (bypasses orchestration rules with force=True):

import asyncio
from prefect.client.orchestration import get_client
from prefect.states import Completed

FLOW_RUN_ID = "YOUR-FLOW-RUN-ID"

async def main():
    async with get_client() as client:
        result = await client.set_flow_run_state(
            flow_run_id=FLOW_RUN_ID,
            state=Completed(message="Manually marked completed after server error"),
            force=True,  # bypass orchestration rules
        )
        print(result.status, getattr(result, "details", None))

asyncio.run(main())

- If you just need it to stop showing as running (not ideal if it really completed), you can cancel it:

prefect flow-run cancel <FLOW_RUN_ID>

Investigate the root cause (why the server errored) The “server throws error on Completed” usually comes from one of these: - Database issues during state write (integrity errors, schema mismatch, database locks — especially with SQLite) - Pending migrations after an upgrade - Rarely, a bug in orchestration or a failing side-effect (e.g., notification/automation) Do this checklist: 1) Inspect the run and recent states

prefect flow-run inspect <FLOW_RUN_ID> -o json

2) Check server logs at the exact time the state transition was attempted - Look for phrases like “Encountered error during state validation”, IntegrityError, OperationalError, “database is locked”, or anything from SQLAlchemy/asyncpg. - If you’re on Docker/systemd, pull recent logs and grep for errors around the timestamp of the run finishing. 3) If you recently upgraded Prefect, run database migrations

prefect server database upgrade

Then restart your server process and retry a new flow run. 4) If you’re on SQLite and see “database is locked” - Restart the server to release locks, then consider migrating to Postgres for reliability with concurrent writes. 5) If you have notification policies/automations that trigger on Completed - Temporarily disable them to rule out a failing side-effect (we can dig into this if you share what you’ve configured). Helpful diagnostics (optional) - Last few states for the run (sometimes shows a “Rejected/Aborted” detail):

import asyncio
from prefect.client.orchestration import get_client

FLOW_RUN_ID = "YOUR-FLOW-RUN-ID"

async def main():
    async with get_client() as client:
        states = await client.read_flow_run_states(FLOW_RUN_ID)
        for s in states[-5:]:
            print(s.name, s.type.value, s.timestamp, s.message)

asyncio.run(main())

Next steps - If you paste the server error snippet plus whether you’re on SQLite/Postgres, I can pinpoint the fix (migrations, constraint cleanup, lock mitigation, etc.). - If you’re blocked right now, run the “force Completed” snippet above to clear the stuck run, then we’ll address the underlying cause to prevent recurrence.

Short answer: a run stuck in Running will still hold a concurrency slot; it only holds CPU/memory if the underlying process/container/pod is still alive. If the worker’s process already exited, you’re not burning infra resources—just a slot that can block other runs until it’s released. How to check quickly - Inspect the run and see if there’s an infrastructure PID (and other details):

prefect flow-run inspect <FLOW_RUN_ID> -o json

- If you see an

infrastructure_pid

and your worker is still active, that process/container/pod may still be running. If it’s empty or the infra is already gone, you’re mainly just holding a concurrency slot. Free resources/slots now - Preferred: cancel the run (this tells the worker to kill infra and releases the slot on terminal state):

prefect flow-run cancel <FLOW_RUN_ID>

- If the server is rejecting state changes and cancel doesn’t clear it, force a terminal state (releases slots immediately):

import asyncio
from prefect.client.orchestration import get_client
from prefect.states import Cancelled, Completed

FLOW_RUN_ID = "YOUR-FLOW-RUN-ID"

async def main():
    async with get_client() as client:
        # Pick one based on what actually happened
        new_state = Cancelled(message="Manual recovery from stuck Running")
        # new_state = Completed(message="Work finished; manually marking Completed")

        result = await client.set_flow_run_state(
            flow_run_id=FLOW_RUN_ID,
            state=new_state,
            force=True,  # bypass orchestration rules
        )
        print(result.status, getattr(result, "details", None))

asyncio.run(main())

- If you do nothing: Prefect’s lease mechanism will auto-release held slots after their TTL (commonly ~5 minutes) plus a short background sweep. That won’t reclaim CPU/memory if the infra is actually still running; for that, use cancel (or delete the pod/container directly if you know it). Good next steps - If you’re worried the process/container is still alive, run the cancel command above. For K8s/Docker, you can also terminate the job/container directly if you can’t get the state to change. - If you can, share: - Prefect version (assuming 3.x) - Cloud or self-hosted, and which worker type (Subprocess/Docker/Kubernetes) - The server log error that appeared when it tried to mark Completed I can then help pinpoint why the state change failed so it doesn’t recur.

Thanks — that detail helps a lot. What’s happening - Your flow finished and tried to set its final state, but the server returned 500 on the set_state endpoint. The runner then tried to mark it Crashed and hit the same 500. - The Kubernetes job/process has already exited (exit code 1 in your logs), so you are not burning CPU/memory in the cluster now. The run is just stuck in “Running” in Prefect and may be holding a concurrency slot until it’s terminal or the lease expires. Unstick this run now - Try a normal cancel first (releases slots and ensures any lingering infra is killed if still present):

prefect flow-run cancel c73ea12b-773d-4cef-b7ef-8361e7b97354

- If that doesn’t change the state because the server still 500s, force a terminal state (releases slot immediately):

import asyncio
from prefect.client.orchestration import get_client
from prefect.states import Cancelled, Completed

FLOW_RUN_ID = "c73ea12b-773d-4cef-b7ef-8361e7b97354"

async def main():
    async with get_client() as client:
        state = Cancelled(message="Manual recovery from stuck Running")
        # Or use Completed(...) if you know it finished successfully
        result = await client.set_flow_run_state(flow_run_id=FLOW_RUN_ID, state=state, force=True)
        print(result.status, getattr(result, "details", None))

asyncio.run(main())

Note: If the server is throwing 500s inside its state write path, even the force call may fail — then proceed to “Fix the server 500” below. Double-check whether anything is actually running - Check for any leftover job/pods for this run:

kubectl get pods -A -l "<http://prefect.io/flow-run-id=c73ea12b-773d-4cef-b7ef-8361e7b97354|prefect.io/flow-run-id=c73ea12b-773d-4cef-b7ef-8361e7b97354>"

- If you see a job/pod still running, you can remove it:

kubectl delete job -A -l "<http://prefect.io/flow-run-id=c73ea12b-773d-4cef-b7ef-8361e7b97354|prefect.io/flow-run-id=c73ea12b-773d-4cef-b7ef-8361e7b97354>"

If you use concurrency limits - The slot will auto-release when the run reaches a terminal state or when the lease expires (commonly ~5 minutes by default), whichever happens first. For immediate relief, use cancel or force a terminal state as above. Fix the server 500 (root cause) On self-hosted Kubernetes, the 500 almost always comes from the API server trying to persist the state and failing (DB/migration/locking). Please gather this: 1) Server API logs around the timestamp of the error - Identify the API pod:

kubectl get pods -n <your-namespace> -l "<http://app.kubernetes.io/name=prefect-server|app.kubernetes.io/name=prefect-server>"

- Get recent logs and search for the failure around set_state:

kubectl logs -n <your-namespace> <api-pod-name> --tail=500

Look for errors like IntegrityError, OperationalError, “database is locked”, “asyncpg…”, or a stack trace from the state validation/flush. 2) Confirm your database - Are you using Postgres or SQLite? Running Prefect Server on Kubernetes with SQLite is prone to write-locks and 500s; Postgres is strongly recommended. - If SQLite: move to Postgres and re-deploy the server pointing to it via

PREFECT_DATABASE_CONNECTION_URL

. This alone resolves most k8s 500-on-set_state issues. 3) Run database migrations (especially after upgrades) - Run this once (with the same image/env as your server) as a Job or by exec’ing into the API pod:

prefect server database upgrade

- Then restart the API pod(s):

kubectl rollout restart deploy/<your-prefect-server-deployment> -n <your-namespace>

4) Ensure only one API replica if you’re on SQLite - If you can’t switch to Postgres immediately, keep the API at 1 replica to avoid concurrent writes/locks. 5) Re-test a new short flow - Once logs are clean and migrations applied, run a tiny flow to confirm set_state works end-to-end. Optional guardrails - Enable flow heartbeats and an automation to mark zombie flows Crashed when heartbeats stop; this prevents long-lived stuck “Running” states if the infra exits: - Set

PREFECT_FLOWS_HEARTBEAT_FREQUENCY

(e.g., 30) on your flow runs

- Configure an automation to mark runs Crashed when no heartbeat is seen within N seconds If you can share: - Which DB you’re using (Postgres vs SQLite) - A snippet of the server API error stack trace around the 500 I can pinpoint the exact fix (migrations, DB lock, constraint issue) and help you make it permanent so this doesn’t recur.

In Prefect 3.x with the Kubernetes worker, each Job/Pod is labeled so you can find it by flow run. Look up the job/pod by flow-run-id - Prefect sets these labels on the Job and its Pod(s): -

<http://prefect.io/flow-run-id=<FLOW_RUN_ID|prefect.io/flow-run-id=<FLOW_RUN_ID>>

-

<http://prefect.io/flow-run-name=<FLOW_RUN_NAME|prefect.io/flow-run-name=<FLOW_RUN_NAME>>

- Also useful:

<http://prefect.io/deployment-id|prefect.io/deployment-id>

,

<http://prefect.io/deployment-name|prefect.io/deployment-name>

,

<http://prefect.io/flow-id|prefect.io/flow-id>

,

<http://prefect.io/flow-name|prefect.io/flow-name>

,

<http://prefect.io/work-pool-name|prefect.io/work-pool-name>

,

<http://prefect.io/worker-name|prefect.io/worker-name>

(present when applicable) Try:

# Jobs
kubectl get jobs -A -l "<http://prefect.io/flow-run-id=c73ea12b-773d-4cef-b7ef-8361e7b97354|prefect.io/flow-run-id=c73ea12b-773d-4cef-b7ef-8361e7b97354>" -o wide

# Pods (labels propagate from the job's pod template)
kubectl get pods -A -l "<http://prefect.io/flow-run-id=c73ea12b-773d-4cef-b7ef-8361e7b97354|prefect.io/flow-run-id=c73ea12b-773d-4cef-b7ef-8361e7b97354>" -o wide

# If many namespaces, the -A flag searches all

If you don’t have the flow-run-id handy but know the name, swap the selector:

kubectl get jobs -A -l "<http://prefect.io/flow-run-name=<your-flow-run-name>|prefect.io/flow-run-name=<your-flow-run-name>>"
kubectl get pods -A -l "<http://prefect.io/flow-run-name=<your-flow-run-name>|prefect.io/flow-run-name=<your-flow-run-name>>"

From Prefect UI/CLI to K8s -

prefect flow-run inspect <FLOW_RUN_ID> -o json

will often show

infrastructure_pid

in the form

namespace:job_name

. You can then:

kubectl -n <namespace> get job <job_name> -o yaml
kubectl -n <namespace> get pods -l job-name=<job_name>

If you still can’t find it - The job/pod may already have terminated/been cleaned up by Kubernetes. In that case, nothing is currently consuming cluster resources even though Prefect shows “Running”. - To ensure cleanup and release any held concurrency slot, cancel or force-complete/cancel the run as shown earlier. Want me to help search live? If you can share the namespace(s) your worker uses and either the flow run id or name, I can suggest the exact

kubectl

commands to copy/paste.

“Cancelling” means Prefect has registered your cancel request but is waiting for the worker to actually stop the Kubernetes Job/Pod and then set the final CANCELLED state. If it sits in Cancelling, one of these is usually true: - The Kubernetes worker can’t kill the Job/Pod (RBAC/permissions, wrong namespace, bad infrastructure_pid). - The Job/Pod is stuck terminating (finalizers, long terminationGracePeriodSeconds, volume detach, etc.). - The Job/Pod is already gone but the server can’t write the final CANCELLED state (you’ve seen 500s on set_state). - The worker isn’t seeing/processing the cancel request (worker offline/misconfigured). Work through this checklist: 1) Confirm whether the Job/Pod is still running - Find it by flow-run-id:

kubectl get jobs -A -l "<http://prefect.io/flow-run-id=<FLOW_RUN_ID>|prefect.io/flow-run-id=<FLOW_RUN_ID>>" -o wide
kubectl get pods -A -l "<http://prefect.io/flow-run-id=<FLOW_RUN_ID>|prefect.io/flow-run-id=<FLOW_RUN_ID>>" -o wide

- Or get the infrastructure pid and jump straight to it:

prefect flow-run inspect <FLOW_RUN_ID> -o json | jq '.infrastructure_pid'
# Expect "namespace:job_name"
kubectl -n <namespace> get job <job_name> -o yaml
kubectl -n <namespace> get pods -l job-name=<job_name>

2) If the Job/Pod is present, try deleting it manually - Foreground delete waits for pods to terminate (what the worker does under the hood):

kubectl -n <namespace> delete job <job_name> --cascade=foreground --wait=true

- If pods hang in Terminating, describe them for clues:

kubectl -n <namespace> describe pod <pod_name>
kubectl -n <namespace> get events --sort-by=.lastTimestamp

- If you must, force delete a stuck pod:

kubectl -n <namespace> delete pod <pod_name> --grace-period=0 --force

After the Job/Pod is gone, Prefect should move from Cancelling → Cancelled. If it doesn’t, see step 4. 3) Check the worker logs for cancellation handling - Look for messages like “Received cancel” / “Killing infrastructure” / errors:

kubectl logs -n <worker-namespace> deploy/<your-prefect-kubernetes-worker-deployment> --tail=500 | grep -i -E "cancel|kill|infrastructure|forbidden|not found|error"

Common problems you’ll see: - RBAC forbidden on jobs/pods delete → the worker can’t delete the job. - NotFound or wrong namespace → the worker’s

infrastructure_pid

points somewhere it can’t reach. - API errors from the K8s API. 4) If the Job/Pod is already gone but Prefect is still “Cancelling” - This likely hits the same server 500 you saw earlier when setting terminal states. Try forcing the final state:

import asyncio
from prefect.client.orchestration import get_client
from prefect.states import Cancelled

async def main():
    async with get_client() as client:
        result = await client.set_flow_run_state(
            flow_run_id="<FLOW_RUN_ID>",
            state=Cancelled(message="Finalizing after manual job deletion"),
            force=True,
        )
        print(result.status, getattr(result, "details", None))

asyncio.run(main())

- If that call 500s, fix the server first (apply DB migrations, ensure Postgres not SQLite in k8s, check API logs for IntegrityError/OperationalError). Once the server can write states again, re-run the force call. 5) Verify the worker’s permissions (RBAC) The worker’s ServiceAccount must be able to delete Jobs and Pods. Example ClusterRole/Binding:

apiVersion: <http://rbac.authorization.k8s.io/v1|rbac.authorization.k8s.io/v1>
kind: ClusterRole
metadata:
  name: prefect-worker
rules:
- apiGroups: ["batch"]
  resources: ["jobs"]
  verbs: ["get","list","watch","create","patch","delete"]
- apiGroups: [""]
  resources: ["pods","pods/log"]
  verbs: ["get","list","watch","delete","patch"]
---
apiVersion: <http://rbac.authorization.k8s.io/v1|rbac.authorization.k8s.io/v1>
kind: ClusterRoleBinding
metadata:
  name: prefect-worker
roleRef:
  apiGroup: <http://rbac.authorization.k8s.io|rbac.authorization.k8s.io>
  kind: ClusterRole
  name: prefect-worker
subjects:
- kind: ServiceAccount
  name: <worker-service-account>
  namespace: <worker-namespace>

6) Termination grace and finalizers

- If your Job template or Pod has a large

terminationGracePeriodSeconds

, Cancelling will last until that grace elapses. - Finalizers on pods or stuck volume detach can also hold termination open; remove blocking finalizers only if you understand the impact. What to send me to pinpoint the cause - Output of: -

prefect flow-run inspect <FLOW_RUN_ID> -o json | jq '{state: .state.name, infra_pid: .infrastructure_pid}'

-

kubectl get jobs -A -l "<http://prefect.io/flow-run-id=<FLOW_RUN_ID>|prefect.io/flow-run-id=<FLOW_RUN_ID>>" -o name

-

kubectl get pods -A -l "<http://prefect.io/flow-run-id=<FLOW_RUN_ID>|prefect.io/flow-run-id=<FLOW_RUN_ID>>" -o wide

- Any error lines from the worker logs around the cancel attempt - Whether your Prefect API can now write terminal states (did a small test run complete/cancel cleanly?) Once the Job/Pod is actually deleted and the server can write states, the run will move from Cancelling to Cancelled and release any held concurrency slots.

@Marvin in what state a flow occupy concurrency limit slot? I see there are 213 flows stuck in cancelling state

Short answer - Slots are held in PENDING, RUNNING, and CANCELLING. - Slots are released only when the run leaves those states for a terminal state (COMPLETED, FAILED, CANCELLED, CRASHED) or is moved back to a non-slot-holding state (e.g., PENDING→SCHEDULED due to rejection). So, 213 runs stuck in CANCELLING will keep occupying deployment concurrency slots until they reach a terminal state or their leases expire and are reaped by the server’s repossessor. What to do now - Prefer to finish cancellation by ensuring the Kubernetes Job/Pod is actually deleted; then Prefect can move CANCELLING → CANCELLED and release the slot: 1) Find K8s resources by label:

kubectl get jobs -A -l "<http://prefect.io/flow-run-id=<FLOW_RUN_ID>|prefect.io/flow-run-id=<FLOW_RUN_ID>>"
  kubectl get pods -A -l "<http://prefect.io/flow-run-id=<FLOW_RUN_ID>|prefect.io/flow-run-id=<FLOW_RUN_ID>>"

2) Delete the Job (foreground) so pods terminate cleanly:

kubectl -n <ns> delete job <job_name> --cascade=foreground --wait=true

3) If pods hang in Terminating, force delete the pod to unblock:

kubectl -n <ns> delete pod <pod_name> --grace-period=0 --force

- If the Kubernetes resources are already gone but runs are still CANCELLING: - You’re likely hitting the same server 500 when writing the final state. Fix the server first (migrations/DB), then force the terminal state to free slots:

import asyncio
  from prefect.client.orchestration import get_client
  from prefect.states import Cancelled

  async def main():
      async with get_client() as client:
          result = await client.set_flow_run_state(
              flow_run_id="<FLOW_RUN_ID>",
              state=Cancelled(message="Finishing stuck CANCELLING run"),
              force=True,
          )
          print(result.status, getattr(result, "details", None))

  asyncio.run(main())

- If you need an immediate bulk relief (nuclear option): - Temporarily reset the affected concurrency limit’s active slots to 0 via the server API or database. This frees capacity but should be used carefully because it ignores which specific runs were holding slots. Prevention tips - Ensure the worker can reliably delete jobs/pods (RBAC, correct namespaces). - Keep terminationGracePeriodSeconds reasonable and avoid blocking finalizers. - Use Postgres (not SQLite) for Prefect Server on Kubernetes and run

prefect server database upgrade

after upgrades. - Consider enabling heartbeats + an automation to mark zombie flows Crashed if the process disappears, so they don’t linger in RUNNING/CANCELLING. If you share one stuck FLOW_RUN_ID and the current K8s namespace used by your worker, I can give you exact kubectl commands to clear it and verify the slot releases.

Yes. In Prefect 3.x, Cancelling still holds concurrency slots. - Slot-holding states: PENDING, RUNNING, and CANCELLING - Slots are released only when the run leaves those states for a terminal state (COMPLETED, FAILED, CANCELLED, CRASHED) or is moved back to a non-slot state - This applies to both deployment (global) concurrency limits and tag-based task concurrency If your runs are stuck in Cancelling, those slots won’t free until: - The Kubernetes Job/Pod is actually terminated and the run reaches a terminal state, or - You force a terminal state (e.g., Cancelled), or - The lease expires and the repossessor reclaims it (default TTL is typically a few minutes, configurable) Given your server has been returning 500s on set_state, you may need to: - Ensure the Job/Pod is deleted (so cleanup is done), then - Fix the server error (DB/migrations) and force the state to CANCELLED to release the slots immediately.

Short answer - Flow/deployment concurrency: deleting the flow run in Cancelling will free the deployment’s active slot count immediately. - Task tag-based concurrency: deleting the task run in Cancelling does not trigger the normal release; the tag slots are not freed immediately and will only free when the lease expires (TTL) or you manually reset the affected tag limit. Why - Slot release is normally done by orchestration rules when a run transitions out of PENDING/RUNNING/CANCELLING to a terminal state. Deletion bypasses those rules. - For flows, Prefect runs a cleanup step on delete that decrements the deployment’s active slot count. The associated lease record may linger until its TTL expires, but it won’t block new runs. - For tasks, deletion does not run the tag-based concurrency cleanup. Those tag leases are not revoked by deletion, so capacity can remain “stuck” until the lease TTL expires and the background reaper reclaims it. If TTLs are long, that looks like the slots are lost. Recommended actions - Prefer setting a terminal state over deleting: - Force the task run to Cancelled/Failed/Completed; that triggers immediate release of tag-based slots. - If you already deleted task runs and need capacity now: - Wait for lease TTL to expire (background service will reclaim). - Or manually reset the specific tag limit (in the UI’s Concurrency page or via API) to set active slots back to 0. Use caution—this resets the limit for all holders of that tag. If you want, share one affected tag name and I’ll outline the exact reset steps for your environment.

Thanks for the snippet — that “Created task run …” followed by no progress is a classic sign the task is waiting to acquire a concurrency slot. What’s happening - Your task uses:

async with concurrency("extraction_service", occupy=1):
    data = await extraction_service.extract_text(file_payload)

- That context manager requests a tag-based concurrency slot named

extraction_service

. If the limit for that name is at capacity (active_slots == limit), the call blocks and silently retries until a slot frees up. - By default, it waits indefinitely (no timeout) and keeps retrying. Confirm it - List your global (tag-based) limits and see if

extraction_service

is at capacity:

prefect global-concurrency-limit ls
prefect global-concurrency-limit inspect extraction_service

Look for active_slots == limit. If so, the task is simply queued on the limit. Options to unblock or tighten behavior - Free capacity now - Wait for other holders to finish, or - Temporarily raise the limit:

prefect global-concurrency-limit update extraction_service --limit <NEW_LIMIT>

- Or temporarily disable it:

prefect global-concurrency-limit disable extraction_service

- Add a timeout/retry policy to avoid “waiting forever”:

async with concurrency(
    "extraction_service",
    occupy=1,
    timeout_seconds=300,   # total time to wait for a slot
    max_retries=100,       # optional: cap number of retries
    # strict=True,         # optional: raise if limit not configured
):
    ...

If

timeout_seconds

elapses without a slot, the context raises a TimeoutError so your task can fail or retry instead of appearing stuck. - If you don’t actually have a limit configured for this name and want to catch misconfigurations fast, set

strict=True

. Without strict, missing limits are treated as “no limit” and won’t block. Extra diagnostics - Turn on debug logs on the worker to see the backoff messages while it waits:

PREFECT_LOG_LEVEL=DEBUG

You’ll see lines like “Unable to acquire concurrency slot… Retrying in X second(s).” - In the UI, check the Concurrency page for

extraction_service

to see holders and recent activity. If you share the output of: -

prefect global-concurrency-limit inspect extraction_service

I can suggest the quickest/safest change to get these tasks moving.

Yes—manually reducing active_slots below the limit will immediately free capacity and unblock tasks that are waiting on that global (tag-based) concurrency limit. It’s an administrative override. How to do it - Inspect the limit:

prefect global-concurrency-limit inspect extraction_service

- Reset the counter (safe when you know holders are stale/zombie):

prefect global-concurrency-limit update extraction_service --active-slots 0

- Alternative quick-unblock (disables enforcement temporarily):

prefect global-concurrency-limit disable extraction_service
# … let the backlog drain …
prefect global-concurrency-limit enable extraction_service

- Or just raise the limit briefly:

prefect global-concurrency-limit update extraction_service --limit <NEW_LIMIT>

Caveats - Use this when active_slots is inflated by stale holders (e.g., runs stuck in Cancelling, deleted tasks that didn’t release, previous server 500s). If there are still genuinely running tasks holding leases, resetting active_slots will allow over-commit relative to your intended limit. - Best practice is to release slots by finishing cancellation or forcing a terminal state on the offending runs. The manual reset is a fast relief valve while you fix the root cause. - If you frequently see this, add a timeout in your code so tasks don’t wait forever:

async with concurrency("extraction_service", timeout_seconds=300):
    ...

If you share the current inspect output for extraction_service, I can recommend whether to reset active_slots to 0, bump the limit, or disable/enable briefly.

You can add a timeout to the concurrency acquisition by setting timeout_seconds on the concurrency context. If a slot isn’t acquired within that time, it raises TimeoutError so your task won’t sit “forever.” Example (async task, with retry on timeout via your task’s retries):

from prefect.concurrency.asyncio import concurrency

@task(name="Extract Text Task", retries=1, retry_delay_seconds=1)
async def extract_text_from_file(kb: KnowledgeBase, file_bytes: dict[str, Any]) -> dict[str, Any]:
    app_config: AppConfig = get_app()
    logger: logging.Logger = app_config.logger
    extraction_service: ExtractionService = app_config.service_registry.get(ExtractionService.__name__)

    file_payload: tuple[str, bytes, str] = [
        ("files", (kb.name, file_bytes, kb.metadata.get("content_type"))),
    ]

    try:
        # Adjust timeout_seconds to how long you're willing to wait for a slot
        async with concurrency(
            "extraction_service",
            occupy=1,
            timeout_seconds=300,   # e.g., 5 minutes
            max_retries=None,      # optional: cap retries; None = retry until timeout
            # strict=True,         # optional: fail fast if the limit isn't configured
        ):
            data = await extraction_service.extract_text(file_payload)
    except TimeoutError:
        logger.warning("Timed out waiting for 'extraction_service' concurrency slot for KB %s", kb.id)
        raise
    except ExtractionServiceException as e:
        logger.error("Task failed for KB %s: %s", kb.id, e.message)
        raise

    return data[0]

Notes - timeout_seconds is the total wall-clock time to wait. If reached, the context raises TimeoutError and your task will fail → Prefect will retry based on your task’s retries. - max_retries caps the number of acquisition attempts; you can leave it None and rely on timeout_seconds. - strict=True is optional: it raises immediately if the named limit doesn’t exist (useful to catch misconfigurations).

Short answer: you’re right. In Prefect 3.x, you can’t move a CANCELLED flow run back to execution without either: - Forcing a state change on the existing run, or - Creating a brand new run from the deployment How to fix your automation: Option A — Force-reschedule the same run (requires a deployment on that run) - Use the “Change flow run state” action - Set state to SCHEDULED (not PENDING) - Enable force Example (YAML-style):

actions:
  - type: "change-flow-run-state"
    state: "SCHEDULED"
    force: true
    message: "Rescheduled by automation after cancellation"

Notes: - This only works if the cancelled run is associated with a deployment - CANCELLED → PENDING is blocked; use SCHEDULED - With force=true, Prefect bypasses the normal terminal-state guardrails Option B — Recommended: create a new run - Use the “Run deployment” action to create a fresh flow run - No force flag needed; works regardless of the original run’s state Example:

actions:
  - type: "run-deployment"
    source: "selected"
    deployment_id: "<DEPLOYMENT_ID>"
    # optional:
    # parameters: {...}
    # schedule_after: "0s"

Quick troubleshooting checklist: - If you used “Change flow run state”, did you select SCHEDULED and set force=true? - Is the cancelled run actually tied to a deployment? If not, use “Run deployment” - Is your automation enabled and filtering the right events (e.g., matches “prefect.flow-run.Cancelled” for the correct flow/deployment)? - Is a worker online in the target work pool to pick up SCHEDULED runs? If you can share your automation config (screenshot or YAML), I can pinpoint the exact tweak needed.