Question regarding result persistence and restarting failed flows: I have a flow that resembles this that I am running locally and on kubernetes

@flow
def demo_flow(x=1):
    val = task_1(x)
    val = task_2(val)
    val = task_3(val)
    val = task_4(val)
    val = prone_to_fail_task(val)
    return val

If expensive_prone_to_fail_task fails, I want to be able to restart my flow and have it retry just

prone_to_fail_task

again. What do I need to achieve that? Will adding

cache_key_fn=task_input_hash,

to just

task_4

be sufficient?

I think you need to configure your result storage appropriately See the relevant docs page here https://docs.prefect.io/concepts/results/#result-storage-location