@Marvin how can I use retry for a crashed run? I wanna set up retry mechanism when the flow run went into "crashed" status only

To set up a retry mechanism for a Prefect flow when the run status goes into "crashed," you can leverage Prefect's retry capabilities. Here's a concise guide for both Prefect 2.x and 3.x versions: Prefect 2.x: In Prefect 2.x, you can use the

retries

and

retry_delay_seconds

arguments within your flow or task definition. While there isn't a direct configuration for retrials based solely on the "crashed" status, implementing generic retrials can allow your flow to handle unexpected interruptions: - Here’s a basic example of adding retries:

python
from prefect import flow

@flow(retries=3, retry_delay_seconds=5)
def my_flow():
    # Your flow logic here

Unfortunately, specific handling for the "crashed" status would require manual monitoring or hooks. Explore the troubleshooting guide for Prefect 2.x for more insights. Prefect 3.x: In Prefect 3.x, there are more advanced options. You can use a

retry_condition_fn

to customize when a task should be retried, potentially allowing you to define conditions based on the task state: - More from Prefect 3.x: task and retry guidelines. Retrial functionality doesn’t directly monitor a "crashed" status; however, integrating condition functions or external process monitoring should help. Remember, though, I'm just an android, thinking anything rarely turns out the way you want it to. For your specific requirement, integrating state change listeners or customizing a monitoring process to relaunch a new flow on crash might be necessary.