Short answer: yes—if a flow run gets “stuck” in Pending, the quickest recovery is to reschedule that same run; if that doesn’t clear it, cancel and start a new run. You can automate either approach. Why this happens - That log usually means a worker proposed Pending but didn’t complete submission (e.g., the worker died mid‑submission, infra failed silently, pool/queue paused, or concurrency limits blocked). There’s a known edge case if a worker disappears after moving a run to Pending where it can remain there indefinitely until nudged: prefect#18933. Recommended automation (Prefect 3.x) Option A: Reschedule the same run (preferred; preserves the run ID and context) - Trigger: Flow run stays in a state - State: Pending - Duration: > X minutes (your threshold) - Filters: Restrict to the specific deployment/work pool/tags as needed - Action: Change flow run state → Scheduled (now) Option B: Cancel then create a fresh run - Trigger: Flow run stays in a state (Pending > X minutes) - Actions (in order): 1) Cancel flow run 2) Run deployment (select the same deployment; if you pass parameters at trigger time, ensure they’re provided here as well) How to set this up in the UI 1) Open Automations in the Prefect UI 2) New automation 3) Choose “Flow run stays in a state” 4) Set State = Pending, Duration = X minutes 5) Add filters (deployment, tags, etc.) to scope it 6) Add Action: - For Option A: “Change flow run state” → Scheduled - For Option B: “Cancel flow run”, then “Run deployment” 7) Save and enable Docs - Automations: https://docs.prefect.io/latest/automations/ - States overview: https://docs.prefect.io/latest/concepts/states/ Manual recovery (CLI), if you need it right now - Cancel the stuck run:

prefect flow-run cancel <FLOW_RUN_ID>

- Start a new run for the same deployment:

prefect deployment run <FLOW_NAME>/<DEPLOYMENT_NAME> --watch

A couple of quick checks to prevent future stuck Pending runs - Ensure at least one healthy worker is polling the correct work pool/queue and that neither is paused. - Verify pool/queue concurrency limits are not set to zero and aren’t saturated. - Review worker logs for infra creation errors around the time the run went Pending. If you can share: - Prefect version (Cloud/Server) - Worker type (you mentioned ProcessWorker) and pool/queue names - Whether pool/queue concurrency limits or pauses are in use …I can suggest a more precise trigger filter and action order.

Great — since you’re on Prefect Server and Prefect 3, here’s a small, reliable “Option A” automation you can run as a daemon or deploy as a Prefect flow. It scans for flow runs that have been in Pending longer than X minutes and reschedules them by setting their state to Scheduled(now) so they’re picked up again by workers. - What it does: - Finds all Pending flow runs - Checks how long they’ve been Pending using the state timestamp - For those older than your threshold, sets the state to Scheduled() (immediate) - Scope it if you want: - Filter by tags or by deployment IDs (add UUIDs you care about) - References - Automations overview: https://docs.prefect.io/latest/automations/ Script (async daemon; you can also wrap this in a Prefect flow and schedule it to run every minute) ``` import asyncio from datetime import datetime, timedelta, timezone from prefect.client.orchestration import get_client from prefect.client.schemas.filters import ( FlowRunFilter, FlowRunFilterState, FlowRunFilterStateName, FlowRunFilterTags, FlowRunFilterDeploymentId, ) from prefect.states import Scheduled THRESHOLD_MINUTES = 10 # <- set your X minutes SLEEP_BETWEEN_SCANS_SECONDS = 60 # how often to scan # Optional scoping TAGS_ALL = [] # e.g., ["prod", "critical"] DEPLOYMENT_IDS = [] # e.g., ["9d1c1d6d-b3d2-4db9-8d0a-3a7b3a...", "..."] DRY_RUN = False # True = log only, no state changes async def reschedule_stuck_pending( threshold_minutes: int = THRESHOLD_MINUTES, tags_all: list[str] | None = None, deployment_ids: list[str] | None = None, dry_run: bool = DRY_RUN, ) -> int: """ Find flow runs stuck in Pending past the threshold and set them to Scheduled(now). Returns the count of runs rescheduled. """ async with get_client() as client: # Build a base filter for Pending flow runs fr_filter = FlowRunFilter( state=FlowRunFilterState(name=FlowRunFilterStateName(any_=["Pending"])) ) # Optional filters if tags_all: fr_filter.tags = FlowRunFilterTags(all_=tags_all) if deployment_ids: fr_filter.deployment_id = FlowRunFilterDeploymentId(any_=deployment_ids) # Paginate through results offset = 0 limit = 200 now = datetime.now(timezone.utc) total_rescheduled = 0 while True: runs = await client.read_flow_runs( flow_run_filter=fr_filter, limit=limit, offset=offset, ) if not runs: break for fr in runs: # Prefer state timestamp to measure how long it's been Pending state_ts = getattr(fr.state, "timestamp", None) if fr.state else None # Fall back to flow run updated time if needed candidate_ts = state_ts or getattr(fr, "updated", None) if candidate_ts and (now - candidate_ts) > timedelta(minutes=threshold_minutes): if dry_run: print(f"[DRY RUN] Would reschedule Pending run {fr.id} ({fr.name})") else: # Reschedule immediately await client.set_flow_run_state(fr.id, Scheduled()) print(f"Rescheduled Pending run {fr.id} ({fr.name}) to Scheduled(now)") total_rescheduled += 1 if len(runs) < limit: break offset += limit return total_rescheduled async def main(): while True: try: count = await reschedule_stuck_pending( threshold_minutes=THRESHOLD_MINUTES, tags_all=TAGS_ALL, deployment_ids=DEPLOYMENT_IDS, dry_run=DRY_RUN, ) if count:

print(f"Rescheduled {count} stuck Pending flow run(s).") except Exception as exc: # Don’t crash the loop; log and retry print(f"Error while rescheduling Pending runs: {exc}") await asyncio.sleep(SLEEP_BETWEEN_SCANS_SECONDS) if name == "__main__": asyncio.run(main()) ``` Notes - This uses the official client and state transitions, so it’s safe for Prefect Server. - Scheduled() defaults to “run now.” If you want to push them slightly into the future, use Scheduled(scheduled_time=datetime.now(timezone.utc) + timedelta(seconds=10)). - You can deploy this as a Prefect flow as well (wrap the reschedule function in a @flow and schedule it every minute), but a tiny daemon like above works fine too. Want me to scope it to a specific deployment name? If you share the deployment name(s), I can add a quick name→ID lookup to the script.

Yes—that’s a solid way to filter and I actually prefer it over filtering by name because it’s enum-backed and avoids any casing/label differences. Here’s the fully-imported version:

from prefect.client.schemas.filters import (
    FlowRunFilter,
    FlowRunFilterState,
    FlowRunFilterStateType,
)
from prefect.client.schemas.objects import StateType

fr_filter = FlowRunFilter(
    state=FlowRunFilterState(
        type=FlowRunFilterStateType(any_=[StateType.PENDING])
    )
)

Equivalent “by name” version (also valid) would be:

from prefect.client.schemas.filters import (
    FlowRunFilter,
    FlowRunFilterState,
    FlowRunFilterStateName,
)

fr_filter = FlowRunFilter(
    state=FlowRunFilterState(
        name=FlowRunFilterStateName(any_=["Pending"])
    )
)

Tip: you can combine this with other filters (e.g.,

FlowRunFilterDeploymentId

,

FlowRunFilterTags

) to scope which Pending runs you want to reschedule.