https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> In my flow, I've got code like this. ...
# ask-marvin
m
@Marvin In my flow, I've got code like this. I submit a task and ask for the state. Returning the state according to the source code actually waits for the task to finish.
Copy code
future: State[str | None] = some_task.submit(return_state=True)

    if future.is_failed():
        result = future.result(raise_on_failure=False)
        if isinstance(result, SpecificError):
            return Failed(message=str(result), name="SpecificError")
        return result

    result: str | None = future.result()
However I am getting errors getting the result when the future is not failed, I get errors like this
Copy code
prefect.exceptions.UnfinishedRun: Run is in RUNNING state, its result is not available.
and
Copy code
prefect.exceptions.UnfinishedRun: Run is in PENDING state, its result is not available.
How is this possible if the future is awaited when passing 
return_state=True
? I am using 
prefect-dask
to run tasks
m
✅ thought for 147.7 seconds
return_state=True
on 
task.submit()
does wait — but it also changes what you get back. The key distinction is: - 
some_task.submit()
-> returns a future - 
some_task.submit(return_state=True)
-> returns a State, after waiting So this line is suspicious:
Copy code
future: State[str | None] = some_task.submit(return_state=True)
That variable is not a future anymore; it's a 
State
. Because of that, this logic:
Copy code
if future.is_failed():
        result = future.result(raise_on_failure=False)
        ...
    result = future.result()
is actually calling 
State.result()
, not 
PrefectFuture.result()
. And 
State.result()
will raise 
UnfinishedRun
if the state is not final. Why can that still happen with Dask? With 
prefect-dask
, there's likely a race/edge case where the wrapped Dask future has completed enough for Prefect to hand back a 
State
, but that 
State
may still be 
PENDING
or 
RUNNING
when you immediately call 
.result()
. In other words, the wait happened at the future layer, but the state object you received is not guaranteed to be final in the way your code assumes. So the short answer is: 
return_state=True
waits on the future, but you should not assume the returned 
State
can always immediately yield a result via 
state.result()
under 
prefect-dask
. A safer pattern is to not use 
return_state=True
here. Use the future directly:
Copy code
future = some_task.submit()

    try:
        result: str | None = future.result()
        return result
    except SpecificError as exc:
        return Failed(message=str(exc), name="SpecificError")
If you need to inspect the terminal state, do this instead:
Copy code
future = some_task.submit()
    state = future.wait()

    if state.is_failed():
        result = state.result(raise_on_failure=False)
        if isinstance(result, SpecificError):
            return Failed(message=str(result), name="SpecificError")
        return result

    return future.result()
Or, if you want to stay entirely state-based, guard against unfinished states:
Copy code
state = some_task.submit(return_state=True)

    if not state.is_final():
        return Failed(message=f"Task returned non-final state: {state.type}", name="Unfinished")

    if state.is_failed():
        result = state.result(raise_on_failure=False)
        if isinstance(result, SpecificError):
            return Failed(message=str(result), name="SpecificError")
        return result

    return state.result()
A couple of practical notes: - 
future.wait()
returns a 
State
- 
future.result()
waits for completion and gives you the task result - 
state.result()
only works if the 
State
is final So for your use case, the most robust approach is usually: - use 
submit()
without 
return_state=True
- use 
future.wait()
if you want state inspection - use 
future.result()
for the actual value Relevant references: - Prefect task submission and futures: Tasks - Prefect states: States - `prefect-dask`: prefect-dask If you want, I can help rewrite your exact snippet into a Dask-safe Prefect 3 pattern.
Open in Slack
PreviousNext
Powered by