Short answer: - There isn’t a built-in “bulk retry” button or CLI subcommand in Prefect 3.x. - Yes, you can retry many failed/crashed runs at once by programmatically setting their state back to Scheduled (e.g.,

AwaitingRetry

) via the API/client. - Workers will pick them up according to your work pool/queue configuration, respecting concurrency limits and queue priority. How to retry multiple crashed/failed runs - List targets (optional, for visibility):

prefect flow-run ls --state FAILED --limit 100
prefect flow-run ls --state CRASHED --limit 100

- Minimal async script to reschedule them:

import asyncio
import datetime
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import FlowRunFilter, FlowRunFilterState, FlowFilter
from prefect.states import AwaitingRetry

async def bulk_retry_failed_or_crashed(flow_names=None, limit=200, delay_seconds=0):
    scheduled_time = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(seconds=delay_seconds)

    async with get_client() as client:
        runs = await client.read_flow_runs(
            flow_filter=FlowFilter(name={"any_": flow_names} if flow_names else None),
            flow_run_filter=FlowRunFilter(
                state=FlowRunFilterState(type={"any_": ["FAILED", "CRASHED"]})
            ),
            limit=limit,
        )

        accepted = 0
        for r in runs:
            result = await client.set_flow_run_state(
                flow_run_id=r.id,
                state=AwaitingRetry(scheduled_time=scheduled_time),
                force=True,  # optional; bypasses orchestration checks if needed
            )
            if getattr(result, "status", None) and result.status.value == "ACCEPT":
                accepted += 1
        print(f"Scheduled {accepted} of {len(runs)} runs")

if __name__ == "__main__":
    asyncio.run(bulk_retry_failed_or_crashed(flow_names=None, limit=200, delay_seconds=0))

Notes: - This keeps the same flow run ID and history; you’re asking the orchestrator to try it again. - If you prefer starting fresh runs, submit new ones with:

prefect deployment run <deployment_name>

How workers will schedule them - When you set runs to

AwaitingRetry

(a Scheduled state), they re-enter the queue with a

next_scheduled_start_time

. - Workers poll the work pool/queues and pick up Scheduled runs whose scheduled time has arrived. - Ordering and limits: - Respects work pool and work queue concurrency limits (only so many RUNNING/PENDING at once). - Honors queue priority and the run’s scheduled time. - If you reschedule a large batch, only as many as available “slots” will start; the rest remain Scheduled and get picked up on subsequent polls. Useful knobs - Adjust concurrency if you need more throughput:

prefect work-pool set-concurrency-limit "my-pool" 20

- Consider adding a small delay (e.g., 10–60s) when bulk rescheduling to avoid a thundering herd. - For ongoing automation, use Prefect Cloud Automations to “change flow-run state to AwaitingRetry” when runs fail, or to “run a deployment” on failure. References - States and retries: States - Work pools/queues and workers: Work pools and workers - Concurrency: Concurrency - Automations (Cloud): Automations If you share how you’re selecting the runs (by flow, tag, deployment, or time window), I can tailor the filter snippet for your exact use case.