https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P>, is there a way to retry a failed flo...
# ask-marvin
a
@Marvin, is there a way to retry a failed flow run without creating a new one? I’d like to avoid accumulating multiple runs.
m
✅ thought for 140.3 seconds
Yes. In Prefect 3.x you can retry a failed flow run without creating a new run. Retries keep the same 
flow_run_id
and just increment the run’s 
run_count
(attempt number). Ways to do it: - Configure automatic retries on the flow (same run ID; attempts increment):
Copy code
from prefect import flow

@flow(retries=3, retry_delay_seconds=30)
def my_flow():
    ...
- Manually retry a terminal run from the CLI (same run ID):
Copy code
# By ID or name; worker will pick it up if it's from a deployment
prefect flow-run retry 123e4567-e89b-12d3-a456-426614174000
prefect flow-run retry my-flow-run-name

# If the run is not from a deployment, provide an entrypoint so it can run locally
prefect flow-run retry 123e4567-e89b-12d3-a456-426614174000 --entrypoint ./flows/my_flow.py:my_flow
- Programmatically via the API (set the run back to Scheduled; same run ID):
Copy code
from prefect import get_client
from prefect.states import Scheduled
from datetime import datetime, timezone, timedelta
import asyncio

async def retry_flow_run(flow_run_id: str):
    async with get_client() as client:
        await client.set_flow_run_state(
            flow_run_id=flow_run_id,
            state=Scheduled(
                scheduled_time=datetime.now(timezone.utc) + timedelta(seconds=5)
            ),
            force=True,
        )

asyncio.run(retry_flow_run("123e4567-e89b-12d3-a456-426614174000"))
Notes: - These approaches do not create a new flow run; they reschedule the existing run and increment 
run_count
. - If the run is associated with a deployment, a worker will pick it up when it’s Scheduled; if not, you’ll need to provide an 
--entrypoint
(CLI) or run it locally. Would you like to do this ad-hoc from the CLI, or programmatically? And are your runs coming from deployments or running locally?
a
@Marvin The question I have is: in the UI, I can’t see the retried runs. Is it possible to update the run datetime? Right now, the retry happens inside the same run, so the timestamp stays the same and doesn’t reflect the retry activity in the UI.
m
✅ thought for 351.5 seconds
You’re right: in the UI, retries are grouped into the same flow run, so you won’t see separate “rows” for each attempt. In Prefect 3.x, a retry keeps the same `flow_run_id`; the run’s 
start_time
is set the first time the run enters RUNNING and does not update on subsequent retries. That’s by design, and there isn’t a supported way to manually change the run’s datetime fields (e.g., `created`/`start_time`) to reflect retry activity. What you can see in the UI - Run count: The run’s attempt counter (
run_count
) increments with each retry attempt. - State history timeline: The flow run details page shows every state transition (e.g., AwaitingRetry → Scheduled → Running → …) with timestamps. Hovering or opening state details surfaces when each retry happened. - Logs: Logs include timestamps and can be used to infer attempt timing. Workarounds and tips - Surface retry context in the UI: If you trigger a manual retry programmatically, include a message so it’s obvious in the state history.
Copy code
from prefect import get_client
  from prefect.states import Scheduled
  from datetime import datetime, timezone, timedelta
  import asyncio

  async def manual_retry(flow_run_id: str):
      async with get_client() as client:
          await client.set_flow_run_state(
              flow_run_id=flow_run_id,
              state=Scheduled(
                  scheduled_time=datetime.now(timezone.utc) + timedelta(seconds=5),
                  name="AwaitingRetry",
                  message="Manual retry requested from UI/CLI",
              ),
              force=True,
          )

  asyncio.run(manual_retry("123e4567-e89b-12d3-a456-426614174000"))
- If you truly need a new “row” with a fresh start time in the Runs list, you’ll need to start a new flow run (i.e., don’t use retry). That will create a new 
flow_run_id
with its own `created`/`start_time`. Related docs - How to manually retry a flow run: https://docs.prefect.io/v3/how-to-guides/workflows/retry-flow-runs - Automatic retries (flows/tasks): https://docs.prefect.io/v3/how-to-guides/workflows/retries - REST: Read flow run states (state history): https://docs.prefect.io/v3/api-ref/rest-api/server/flow-run-states/read-flow-run-states - REST: Read a flow run: https://docs.prefect.io/v3/api-ref/rest-api/server/flow-runs/read-flow-run - Flow run schema (fields like 
run_count
, 
start_time
, 
next_scheduled_start_time
): https://docs.prefect.io/v3/api-ref/python/prefect-server-schemas-core If showing per-attempt start times more prominently in the UI would help your team, I’m happy to file a feature request on your behalf.
a
@Marvin Here’s a cleaner and more structured version: I’d like to be able to track retried flow runs in the UI and easily identify them. As mentioned earlier, when a flow run is retried, its original datetime doesn’t change, so it’s not possible to rely on timestamps in the UI to detect retries. I have a retry script that retries runs based on several criteria, and I need a way to identify those runs afterward in the UI. For example, if I retry 5000 runs, I want to be able to filter and see those specific runs a few hours later. One idea would be to add tags during the retry, such as: a tag to indicate that the run was retried (e.g. 
manual-retry
) • a tag including a timestamp (e.g. 
retried-at:<timestamp>
) to identify when the retry batch was triggered This would help answer: 1. How many flow runs were retried? 2. When did the retry happen, and how long did it take (i.e. easily identify the latest batch)?
m
✅ thought for 337.8 seconds
Great idea — tagging retried runs is a clean, supported way to make them discoverable in the UI later. Here’s a recommended pattern that works well at scale. What to add when you retry - A stable marker tag so you can always find retried runs: manual-retry - A batch tag to group a specific retry wave, e.g.: retry-batch:20260420-1500 (UTC timestamp) - Optional: include the same batch tag in the retry state’s message so it shows up clearly in state history How to implement in your retry script - Update the run’s tags, preserving existing tags - Then set the run to Scheduled (AwaitingRetry) with a message that includes the batch tag - Do this for each targeted run Example (async Python, Prefect 3.x)
Copy code
import asyncio
from datetime import datetime, timezone, timedelta
from prefect import get_client
from prefect.states import Scheduled

# Create a batch tag once per execution of your script
BATCH_TAG = f"retry-batch:{datetime.now(timezone.utc).strftime('%Y%m%d-%H%M%S')}"

async def tag_and_retry(flow_run_id: str):
    async with get_client() as client:
        # 1) Add tags
        fr = await client.read_flow_run(flow_run_id)
        existing = set(fr.tags or [])
        new_tags = sorted(existing.union({"manual-retry", BATCH_TAG}))
        await client.update_flow_run(flow_run_id, tags=new_tags)

        # 2) Schedule the retry with a clear message
        msg = f"Manual retry via script ({BATCH_TAG})"
        await client.set_flow_run_state(
            flow_run_id=flow_run_id,
            state=Scheduled(
                scheduled_time=datetime.now(timezone.utc) + timedelta(seconds=5),
                name="AwaitingRetry",
                message=msg,
            ),
            force=True,
        )

async def main(flow_run_ids):
    # Concurrency limit for large batches
    sem = asyncio.Semaphore(50)

    async def worker(fr_id):
        async with sem:
            try:
                await tag_and_retry(fr_id)
            except Exception as e:
                print(f"Failed {fr_id}: {e}")

    await asyncio.gather(*(worker(i) for i in flow_run_ids))

# Supply your target run IDs (from your current criteria)
# asyncio.run(main([...]))
If you prefer raw REST calls - Update tags:
Copy code
PATCH <https://api.prefect.cloud/api/flow_runs/<FLOW_RUN_ID>>
{
  "tags": ["manual-retry", "retry-batch:20260420-1500", "...existing tags..."]
}
- Then set state to Scheduled (AwaitingRetry) with a message:
Copy code
POST <https://api.prefect.cloud/api/flow_runs/<FLOW_RUN_ID>/set_state>
{
  "state": {
    "type": "SCHEDULED",
    "name": "AwaitingRetry",
    "message": "Manual retry via script (retry-batch:20260420-1500)",
    "scheduled_time": "2026-04-20T15:00:05Z"
  },
  "force": true
}
How to find them later - In the UI: on the Runs page, filter by Tags = manual-retry and/or your batch tag (e.g., retry-batch:20260420-1500). This gives you exactly the set you retried. - Via REST to list: https://docs.prefect.io/v3/api-ref/rest-api/server/flow-runs/read-flow-runs Minimal filter body to get only that batch:
Copy code
{
  "flow_runs": {
    "tags": {
      "all_": [
        "retry-batch:20260420-1500"
      ]
    }
  }
}
- Via REST to count: https://docs.prefect.io/v3/api-ref/rest-api/server/flow-runs/count-flow-runs
Copy code
{
  "flow_runs": {
    "tags": {
      "all_": [
        "retry-batch:20260420-1500"
      ]
    }
  }
}
Notes and gotchas - Timestamps won’t change on retry: start_time remains the first RUNNING time. Tags + state history are the reliable way to surface retry activity. - Tags are mutable: they’re great for identification and filtering, but anyone with permissions can remove them later. If auditability is important, also include the batch tag in the retry state message (it’s recorded in state history). - Deployments vs. local runs: setting a run to Scheduled will be picked up by a worker if the run came from a deployment. For non-deployment runs, scheduling alone won’t execute it. Relevant docs
- Manually retry a flow run: https://docs.prefect.io/v3/how-to-guides/workflows/retry-flow-runs - Update a flow run (PATCH /flow_runs/{id}): https://docs.prefect.io/v3/api-ref/rest-api/server/flow-runs/update-flow-run - List/filter flow runs (POST /flow_runs/filter): https://docs.prefect.io/v3/api-ref/rest-api/server/flow-runs/read-flow-runs - Count flow runs (POST /flow_runs/count): https://docs.prefect.io/v3/api-ref/rest-api/server/flow-runs/count-flow-runs If you want, I can also open a feature request to make “retries” a first-class filter in the UI (e.g., run_count > 0 and/or show last retry timestamp) so you wouldn’t need tagging for this.
a
@Marvin I want 
invoice_final_publishing_flow
to run sequentially on a dedicated Prefect worker/process instead of the shared 
process-worker
pool we currently use. Right now, the 
process-worker
deployment is shared across multiple flows and the Kubernetes deployment has multiple worker replicas, so 
invoice_final_publishing
runs in parallel. That creates concurrency issues for this specific workflow. I’d like guidance on the best infrastructure/design approach to guarantee that all runs of 
invoice_final_publishing_flow
are executed strictly one at a time on a single worker instance. I’m thinking this may require: • a dedicated work pool / worker / queue for this flow • a separate Kubernetes deployment with a single replica • deployment changes in Prefect • possibly CI/CD updates as well Could you help identify the cleanest and most reliable setup for this in Prefect, including all infra changes that should be considered?
Copy code
@flow(
    name="invoice-final-publishing-flow",
    log_prints=True,
    timeout_seconds=settings.FLOW_TIMEOUT_SECONDS,
)
async def invoice_final_publishing_flow(file_key: str) -> State:
    """Publish a finalized invoice to the client portal.

    If the invoice already exists in the portal (code 11), it is deleted
    and the upload is retried once.

    Args:
        file_key: S3 key of the invoice zip file.
    """
m
✅ thought for 173.6 seconds
Yes — the cleanest setup is usually: 1. Put this flow on its own dedicated work pool / queue 2. Run a dedicated worker deployment for that pool/queue with exactly 1 replica 3. Set a Prefect concurrency limit of 1 on the deployment 4. Optionally also set the worker process limit to 1 for defense in depth That gives you both: - Prefect-level serialization (only one flow run is allowed at a time) - infra-level isolation (this flow is not competing with your shared worker fleet) For your case, I would not rely on Kubernetes replica count alone. A single replica helps, but the most reliable control is the Prefect deployment concurrency limit = 1. --- ## Recommended architecture ### Best-practice design For 
invoice_final_publishing_flow
, create: - a dedicated work pool or at minimum a dedicated work queue - a dedicated Kubernetes worker deployment that listens only to that pool/queue - worker replicas = 1 - deployment concurrency limit = 1 with enqueue behavior This ensures: - only this flow is routed to that worker - only one run can execute at a time - additional runs wait instead of executing concurrently - shared 
process-worker
traffic cannot interfere --- ## Why both Prefect + infra controls matter ### Prefect concurrency limit = 1 This is the most important part. It ensures only one run of that deployment can be active at once. If multiple runs are scheduled or triggered, they will queue instead of starting together. Without this, even with a dedicated worker, you can still accidentally get overlap if the worker or infra settings change later. ### Dedicated worker / pool / queue This gives operational isolation: - no noisy neighbors from other flows - easier debugging and monitoring - safer future changes to shared worker settings ### Single Kubernetes replica This prevents multiple worker pods from polling the same queue/pool simultaneously. It’s good infra hygiene, but by itself it’s not the full guarantee. The real serialization guarantee should come from Prefect concurrency config. --- ## Cleanest implementation options ## Option A: Dedicated work pool for this flow This is the clearest and easiest to reason about. Use a separate pool such as: - 
invoice-final-publishing-pool
Run one Kubernetes worker deployment against it: - replicas: 
1
Deploy only this flow there. ### Pros - strongest isolation - simple mental model - easiest to audit - avoids accidental routing from other deployments ### Cons - one more pool/worker deployment to manage For a workflow with known concurrency sensitivity, this is usually what I’d recommend. --- ## Option B: Shared work pool + dedicated work queue If you want fewer pools, you can keep the same pool and create a dedicated queue for this flow, then point a single dedicated worker at that queue. ### Pros - less pool sprawl - still provides reasonable isolation ### Cons - slightly easier for future config drift to reintroduce overlap - less clean than a separate pool If this workflow is business-critical and concurrency bugs are painful, I’d prefer Option A. --- ## What I would recommend for your exact situation Because you said: - the current pool is shared - Kubernetes has multiple worker replicas - this flow has known concurrency issues I’d use: Dedicated pool + dedicated worker deployment + deployment concurrency limit of 1 That is the most reliable and maintainable setup. --- ## Infra changes to make ## 1) Create a dedicated work pool Example name: - 
invoice-final-publishing-pool
If you are using Kubernetes workers, make it a Kubernetes-type pool. You can inspect available CLI help with: - 
prefect work-pool --help
I wasn’t able to get CLI help output back from the tool in time, so I won’t guess exact subcommand syntax here. If you want, I can help you craft the exact commands based on your worker type and current deployment method. Relevant docs:
- Work pools - Workers --- ## 2) Create a dedicated Kubernetes worker deployment Create a separate K8s Deployment for this worker with: - 
replicas: 1
- worker pointed only at 
invoice-final-publishing-pool
- optional worker limit set to 
1
Conceptually:
Copy code
bash
prefect worker start --pool invoice-final-publishing-pool --limit 1
Use 
--limit 1
as extra protection so that even that worker only executes one run at a time. ### K8s considerations Also consider: - PodDisruptionBudget if you care about avoiding voluntary eviction - resource requests/limits sized specifically for this flow - anti-autoscaling rules so replicas don’t get increased automatically - separate labels/namespace if you want stronger ops isolation --- ## 3) Update the deployment for this flow Your deployment should target the new pool (or queue) and set concurrency to 
1
. The key requirement is: - all runs of this deployment must be enqueued/serialized - limit = 1 If you deploy from Python, the pattern in Prefect 3 is via 
.deploy(...)
on the flow loaded from source, not old 2.x deployment APIs. For example, conceptually:
Copy code
python
from prefect import flow

@flow(
    name="invoice-final-publishing-flow",
    log_prints=True,
    timeout_seconds=settings.FLOW_TIMEOUT_SECONDS,
)
async def invoice_final_publishing_flow(file_key: str):
    ...

if __name__ == "__main__":
    invoice_final_publishing_flow.deploy(
        name="invoice-final-publishing",
        work_pool_name="invoice-final-publishing-pool",
        concurrency_limit=1,
    )
Depending on how you deploy today, you may instead be using: - 
prefect.yaml
- 
prefect deploy
- 
flow.from_source(...).deploy(...)
Docs: - Deployments - Deployment guides --- ## 4) If using 
prefect.yaml
, isolate this deployment there If your CI/CD uses 
prefect deploy
with 
prefect.yaml
, update only this deployment to point to the dedicated pool and set concurrency to 1. Conceptually, the deployment config should include: - this deployment name - work pool name - optionally work queue name - concurrency limit = 1 If you share your current 
prefect.yaml
, I can help rewrite the exact stanza. --- ## 5) CI/CD changes to consider You likely need to update CI/CD in two places: ### Application deploy pipeline Update deployment registration so this flow deploys to: - the new dedicated pool/queue - concurrency limit 1 ### Infrastructure deploy pipeline Add or modify manifests/Helm/Kustomize/Terraform for: - dedicated worker Deployment - replica count fixed at 1 - worker env vars / Prefect API config - service account / secrets / image / namespace settings ### Also verify - no HPA is attached to this worker deployment - no one reuses the dedicated pool for unrelated flows - worker image contains all dependencies needed by this flow --- ## Important caveat: serialization scope Be clear on what you want serialized: ### If you want: “Only one run of 
invoice_final_publishing_flow
deployment at a time” Then deployment concurrency limit of 1 is perfect. ### If you want: “Only one run across multiple deployments of this same flow” Then you may need a broader concurrency mechanism, such as a shared concurrency control strategy, because deployment-level limits apply per deployment. If you only have one deployment for this flow, then deployment limit = 1 is enough. --- ## What not to rely on alone ### Not enough by itself: single worker replica A single K8s worker pod is helpful, but not sufficient as the only control. ### Not enough by itself: worker 
--limit 1
That protects only that worker process. It does not replace deployment-level concurrency controls as a durable orchestration rule. ### Not ideal: keeping this on the shared pool
Even if you add a limit, the dedicated pool/worker is cleaner because this flow clearly has special execution requirements. --- ## My practical recommendation If this were my setup, I’d do this: 1. Create 
invoice-final-publishing-pool
2. Deploy a dedicated Kubernetes worker to that pool with 
replicas: 1
3. Start the worker with 
--limit 1
4. Move only 
invoice_final_publishing_flow
deployment to that pool 5. Set deployment concurrency limit to 
1
6. Ensure CI/CD owns both the Prefect deployment config and the K8s worker manifest 7. Document that this pool is reserved exclusively for serialized invoice publishing That is simple, explicit, and robust. --- ## Nice extra safeguard If this flow mutates an external system and overlap is especially dangerous, I’d also consider making the flow logic itself idempotent / lock-aware where possible. Reason: infra controls reduce concurrency risk a lot, but application-level protection is still valuable if someone later changes deployment settings by accident. --- ## If you want the minimum-change version If you want the smallest change without introducing a whole new pool: 1. create a dedicated queue under the existing pool 2. run one dedicated worker that polls only that queue 3. set deployment concurrency limit = 1 That can work well, but I still think a dedicated pool is cleaner for a sensitive workflow. --- ## Short answer Best setup: dedicated pool + dedicated single-replica worker + deployment concurrency limit of 1. That is the cleanest and most reliable way to guarantee 
invoice_final_publishing_flow
runs one at a time. If you want, send me either: - your current 
prefect.yaml
, or - how you currently deploy this flow (
prefect deploy
, Python 
.deploy()
, Helm/manifests, etc.) and I can turn this into a concrete migration plan with exact config examples.
5 Views
Open in Slack
PreviousNext
Powered by