<@ULVA73B9P> My flow says Run count: 2 / 1 because...
# ask-marvin
t
@Marvin My flow says Run count: 2 / 1 because it was restarted. How do I disable restarting?
m
thought for 422.3 seconds
Short answer: There isn’t a product toggle to “disable restart” for a flow run in Prefect 3.x. The “Run count: 2 / 1” you’re seeing means: - 2 = this is the second attempt for the same flow run (it was restarted) - 1 = your flow’s retry policy allows 0 retries (1 total allowed attempt via retries), so the extra attempt came from a manual/UI-driven restart, not a retry What you can do instead: - Prevent accidental/manual restarts via permissions (Prefect Cloud): - Use workspace RBAC to restrict who can change flow run states or manage runs. Users without sufficient permissions won’t be able to restart a run from the UI. - See Prefect Cloud workspaces and roles - Make sure automations aren’t restarting flows: - There’s no built-in “Restart” automation action, but an automation could simulate a restart by changing a failed run back to a scheduled/pending state or by running the deployment again. - Review and disable any automations that could re-run failed flows or change flow run states. - Docs: Detect zombie flows and Automations API - Set retries to 0 to avoid automatic re-runs: - Retries are separate from restarts. If your flow has
retries>0
, remove them to prevent automatic retries. - Flow runs API ref (for context on run state/attempts): Flow runs API - Add a defensive guard in your flow (optional): - If you want to hard-block restarts at runtime, you can check the run count and fail early if it’s >1. - Example:
Copy code
from prefect import flow, context

    @flow
    def my_flow():
        ctx = context.get_run_context()
        run_count = getattr(ctx.flow_run, "run_count", 1)
        if run_count > 1:
            raise RuntimeError(f"Restarts disabled for this flow run (run_count={run_count}).")

        # ...rest of your flow...
Questions to help me tailor this: - Do you want to prevent manual restarts by teammates in the UI, or did an automation/CI trigger the second attempt? - Are you using Prefect Cloud (RBAC available) or open-source Server?
t
@Marvin Please help me check max_retries on my flow
m
thought for 127.3 seconds
Happy to help! Quick clarification: do you want to check the retry setting for - a specific flow run that already exists, or - the flow’s code definition (what future runs will use)? Here are both options in Prefect 3.x: - Check for a specific flow run (via client)
Copy code
import asyncio
  from prefect.client.orchestration import get_client

  async def show(flow_run_id: str):
      async with get_client() as client:
          fr = await client.read_flow_run(flow_run_id)
          # max_retries lives on the run's policy
          print("max_retries:", fr.empirical_policy.max_retries)
          print("retry_delay_seconds:", fr.empirical_policy.retry_delay_seconds)

  asyncio.run(show("<your_flow_run_id>"))
Docs: Flow runs API - Check at runtime from inside the flow
Copy code
from prefect import flow, context, get_run_logger

  @flow
  def my_flow():
      logger = get_run_logger()
      ctx = context.get_run_context()
      policy = ctx.flow_run.empirical_policy
      <http://logger.info|logger.info>(f"max_retries={policy.max_retries}, retry_delay_seconds={policy.retry_delay_seconds}")
- Check the flow’s code-defined retries (what the decorator sets)
Copy code
# if you have the Flow object imported
  from my_module import my_flow
  print("flow.retries:", my_flow.retries)
Note: The decorator parameter is `retries`; the server computes
max_retries
for each run from that. If you share a flow_run_id, I can run the client snippet for you (or adapt it) to confirm your current max_retries.