Hi all, I'm seeing a new error crop up in a flow I've been running stably for a few months. The flow is a top-level pipeline that runs a series of

create_flow_run

(and

wait_for_flow_run

) tasks. One of these steps is followed up with a

get_task_run_result

, which has always worked without issue until today. Now it's throwing this error:

ValueError: The task result cannot be loaded if it is not finished

. I'm not seeing how this could be, since I can see in the logs that the upstream task did in fact finish successfully. I tried explicitly setting the result of

wait_for_flow_run

as an upstream dependency of

get_task_run_result

(which i think shouldn't be needed), and I also tried setting the

poll_time

to

30

, but still no luck. Does anyone have any ideas?

can you share the flow structure? do you wait until the child run finishes and did you set the state dependencies correctly to ensure you get the task run result only once all task runs in the child flow run finished?

here's the part that's failing:

flow_etl_trips_id = create_flow_run(flow_name="ETL Trips")
check_flow_etl_trips_id = check_run_terminal_state(
  wait_for_flow_run(flow_etl_trips_id)
)

latest_file_data = get_task_run_result(flow_etl_trips_id, "latest_file-1", poll_time=30)
latest_file_data.set_upstream(check_flow_etl_trips_id)

you may need to raise_final_state to ensure you don't try to retrieve results if the child flow run fails:

from prefect import Flow
from prefect.tasks.prefect import create_flow_run, wait_for_flow_run

PROJECT_NAME = "community"


with Flow("sample_flow") as flow:
    child_run_id = create_flow_run(
        flow_name="child_flow",
        project_name=PROJECT_NAME,
        task_args={"name": "Friendly name for a DAG node"},
    )
    extract_load_wait_task = wait_for_flow_run(
        child_run_id,
        raise_final_state=True,
        stream_logs=True,
        task_args={"name": "Friendly wait task name"},
    )

i'm handling an upstream failure with this (maybe hack-y) task:

@task
def check_run_terminal_state(view):
  if isinstance(
    view.state,
    Success
  ):
    raise signals.SUCCESS()
  else:
    raise signals.FAIL()

I'd recommend sticking to raise_final_state - it already polls for the end state and might be more reliable than manually raising signals

okay, thank you for the suggestion, i'll give that a try!

Yes, you're right I believe upgrading to a more recent version seems like the best way to approach it

okay cool, i'll give that shot. thanks again for your help, i really appreciate it!

That’s really weird. Are you writing out something big? I’m wondering if the task is done, but the checkpoint it still being saved

no, it's just a dict containing one filename and one url

Yeah I can’t think of anything that would cause this if it’s already done unfortunately. I guess something you can do is write it out yourself, and then persist the location in the KV store. And then you can retrieve it. How intermittent is this? Happening every other day? Once a week?

luckily i can safely offload the failing part to one of the sub-flows so i'm just going to re-factor, no big deal. i'd say roughly every other day is right. i have two flows that are impacted though and today one failed but the other succeeded

Ok yeah that’s weird. Will keep this in mind for other reports.

yeah that part i can see is succeeding based on the logs