Yes—there are a few good ways to do this in Prefect 3.x. The key is to avoid creating a direct data dependency on a failed upstream future unless you explicitly allow it. Here are the most common patterns: 1) Allow downstream to run and receive failures with allow_failure - Wrap the upstream future with

allow_failure(...)

. The downstream will run even if the upstream failed (including timeouts), and it will receive the Exception instead of raising

UpstreamTaskError

.

from prefect import flow, task, allow_failure

@task(timeout_seconds=5)
def upstream(i):
    # may timeout or fail
    ...

@task
def downstream(maybe_value):
    if isinstance(maybe_value, Exception):
        # ignore failed upstream
        return None
    return f"processed: {maybe_value}"

@flow
def main():
    fut = upstream.submit(1)          # submit so failure doesn’t stop the flow
    return downstream(allow_failure(fut))

2) Collect successes and ignore failures across many tasks - Submit all tasks, wait for completion, and only materialize successful results. This is great for batching where some items can fail/time out.

from prefect import flow, task
from prefect.futures import wait

@task(timeout_seconds=5)
def work(x: int) -> int:
    # may fail/timeout
    return x * 2

@task
def aggregate(successes: list[int]):
    # do something with only the successful results
    return {"count": len(successes), "values": successes}

@flow
def main():
    items = [1, 2, 3, 4, 5]
    futures = work.map(items)

    done, _ = wait(futures)  # ensures all finish, even if some fail/timeout

    successes = []
    for f in done:
        if f.state.is_completed():
            successes.append(f.result())  # safe after is_completed()

    return aggregate(successes)

3) Get full control by passing states (return_state=True) - You can pass a State to the downstream and decide what to do based on

state.is_completed()

/

state.is_failed()

/

state.name == "TimedOut"

.

from prefect import flow, task

@task(timeout_seconds=5)
def upstream():
    ...

@task
def downstream(up_state):
    if up_state.is_completed():
        return up_state.result()
    if up_state.name == "TimedOut":
        return "timeout handled"
    return "failed handled"

@flow
def main():
    st = upstream.submit(return_state=True)
    st.wait()
    return downstream(st)

Bonus: If you don’t actually need the upstream data, only ordering, use wait_for -

wait_for=[future]

creates an execution dependency but not a data dependency, so downstream will run after upstream completes regardless of outcome. - Don’t pass the upstream future as a parameter if you want to ignore its failure; just use

wait_for

.

@flow
def main():
    a = task_a.submit()
    # Will run after a finishes, even if a fails
    task_b.submit(wait_for=[a])

Quick notes - Timeouts produce a TimedOut (failed) state; all patterns above handle that. - If timeouts are common but transient, consider

retries=

on the upstream tasks. If you can share how you’re wiring the downstream (e.g., mapping over results vs. single downstream aggregator), I can tailor the example directly to your code.

You’re very close. The key change: wrap the inputs to

task_4

with

allow_failure(...)

instead of only using it in

wait_for

. Passing failed futures as data creates a data dependency that will raise

UpstreamTaskError

unless you mark that input as allowed to fail. Simplest working version (ignore failed upstreams, use whatever succeeded):

from prefect import flow, task, allow_failure
from prefect_dask.task_runners import DaskTaskRunner
from dask.distributed import Client

@task
def task_1(a: int, b: int) -> int:
    return a + b

@task
def task_2(c: int) -> int:
    if c == 3:
        raise ValueError("Raised an error on purpose.")
    return c * c

@task
def task_3(d: int) -> int:
    return d + d

@task
def task_4(c, d):
    # c and/or d may be Exception if upstream failed
    c_val = None if isinstance(c, Exception) else c
    d_val = None if isinstance(d, Exception) else d
    print(f"Received c: {c_val}, d: {d_val}")
    # do whatever you want with available values
    return d_val

@flow(log_prints=True)
def main_flow():
    a, b = 1, 2
    future_1 = task_1.submit(a, b)            # -> 3
    future_2 = task_2.submit(future_1)        # will fail on purpose
    future_3 = task_3.submit(future_1)        # will succeed
    # Wrap inputs with allow_failure so task_4 runs and receives Exceptions for failures
    future_4 = task_4.submit(allow_failure(future_2), allow_failure(future_3))
    return future_4

if __name__ == "__main__":
    client = Client(n_workers=2, threads_per_worker=2)
    task_runner = DaskTaskRunner(address=client.scheduler.address)
    result = main_flow.with_options(task_runner=task_runner)().result(raise_on_failure=False)
    print(f"Final result of the flow: {result}")

Notes: - You don’t need

return_state=True

or

wait_for

here.

allow_failure(...)

on the actual inputs lets the downstream run and receive Exceptions for failed upstreams (including timeouts). - If you prefer to inspect states instead, keep

return_state=True

and still wrap with

allow_failure

, then check each input: - If it’s an Exception, treat as failed. - If it’s a State, use

state.is_completed()

/

state.result(raise_on_failure=False)

to extract the value.

Thanks for the details — the error makes sense given the current wiring. What’s happening: -

wait_for=[allow_failure(...)]

only affects orchestration order; it does not change how Prefect resolves the data inputs. - Because

future_2

failed, passing it as a normal argument to

task_4

creates a data dependency that Prefect tries to resolve. Since it’s failed and not marked as allowed-to-fail on the input itself, Prefect raises NotReady/UpstreamTaskError before

task_4

starts. Fix: - Wrap the actual inputs to

task_4

with

allow_failure(...)

. - Drop

return_state=True

here unless you specifically want to pass State objects around (it complicates types and still requires

allow_failure

for failures). Working version with Dask:

from prefect import flow, task, allow_failure
from prefect_dask.task_runners import DaskTaskRunner
from dask.distributed import Client

@task
def task_1(a: int, b: int) -> int:
    return a + b

@task
def task_2(c: int) -> int:
    if c == 3:
        raise ValueError("Raised an error on purpose.")
    return c * c

@task
def task_3(d: int) -> int:
    return d + d

@task
def task_4(c, d):
    # c and/or d may be Exception if upstream failed
    c_val = None if isinstance(c, Exception) else c
    d_val = None if isinstance(d, Exception) else d
    print(f"Received c: {c_val}, d: {d_val}")
    return d_val

@flow(log_prints=True)
def main_flow():
    a, b = 1, 2
    future_1 = task_1.submit(a, b)         # -> 3
    future_2 = task_2.submit(future_1)     # will fail on purpose
    future_3 = task_3.submit(future_1)     # will succeed (-> 6)

    # Critical: wrap the inputs themselves with allow_failure
    future_4 = task_4.submit(allow_failure(future_2), allow_failure(future_3))
    return future_4

if __name__ == "__main__":
    client = Client(n_workers=2, threads_per_worker=2)
    task_runner = DaskTaskRunner(address=client.scheduler.address)
    result = main_flow.with_options(task_runner=task_runner)().result(raise_on_failure=False)
    print(f"Final result of the flow: {result}")

Why this works: -

allow_failure(future_2)

tells Prefect “it’s okay if this input failed; pass the Exception to the task instead of blocking.” - No

wait_for

necessary here; the data dependency already ensures

task_4

won’t run until both upstreams finish. - No `return_state=True`;

task_4

receives either the concrete values or Exception objects, which keeps the types consistent. If you want a version that inspects states instead (e.g., to distinguish TimedOut vs. other failures), we can do that too, but it’s best to avoid mixing

return_state=True

with

allow_failure

unless you’re comfortable handling a mix of State and Exception objects.

Got it — thanks for sticking with this. If

allow_failure

on the direct inputs still isn’t working in your environment, here are two alternative patterns that avoid the NotReady/UpstreamTaskError by resolving values before

task_4

. Option A: Resolve upstream futures in the flow, then pass plain values - This bypasses any failure-propagation on inputs to

task_4

by only passing Python values (or None).

from prefect import flow, task
from prefect.futures import wait
from prefect_dask.task_runners import DaskTaskRunner
from dask.distributed import Client

@task
def task_1(a: int, b: int) -> int:
    return a + b

@task
def task_2(c: int) -> int:
    if c == 3:
        raise ValueError("Raised an error on purpose.")
    return c * c

@task
def task_3(d: int) -> int:
    return d + d

@task
def task_4(c, d):
    print(f"Received c: {c}, d: {d}")
    return d

@flow(log_prints=True)
def main_flow():
    a, b = 1, 2
    f1 = task_1.submit(a, b)

    f2 = task_2.submit(f1)  # will fail
    f3 = task_3.submit(f1)  # will succeed

    # Ensure both finish (some may fail)
    wait([f2, f3])

    # Safely materialize only successful values; use None for failures
    c_val = f2.result() if f2.state.is_completed() else None
    d_val = f3.result() if f3.state.is_completed() else None

    # Now pass plain values to downstream (no upstream failure propagation)
    return task_4.submit(c_val, d_val)

if __name__ == "__main__":
    client = Client(n_workers=2, threads_per_worker=2)
    task_runner = DaskTaskRunner(address=client.scheduler.address)
    result = main_flow.with_options(task_runner=task_runner)().result(raise_on_failure=False)
    print(f"Final result of the flow: {result}")

Option B: Use a small “coerce” task to turn failed states/exceptions into Optional values - This keeps error-handling inside tasks while still avoiding failure propagation into

task_4

.

from prefect import flow, task, allow_failure
from prefect.states import State
from prefect_dask.task_runners import DaskTaskRunner
from dask.distributed import Client

@task
def task_1(a: int, b: int) -> int:
    return a + b

@task
def task_2(c: int) -> int:
    if c == 3:
        raise ValueError("Raised an error on purpose.")
    return c * c

@task
def task_3(d: int) -> int:
    return d + d

@task
def coerce_to_optional(state_or_exc):
    # Accept either a State (successful or failed) or an Exception
    if isinstance(state_or_exc, Exception):
        return None
    if isinstance(state_or_exc, State):
        # Safely get the result without raising
        return state_or_exc.result(raise_on_failure=False)
    # If it's already a plain value
    return state_or_exc

@task
def task_4(c, d):
    print(f"Received c: {c}, d: {d}")
    return d

@flow(log_prints=True)
def main_flow():
    a, b = 1, 2
    f1 = task_1.submit(a, b)

    # Ask upstream tasks to return states explicitly
    s2 = task_2.submit(f1, return_state=True)  # will be a State (failed)
    s3 = task_3.submit(f1, return_state=True)  # will be a State (completed)

    # Convert each to Optional[int] using allow_failure at this step
    c_opt = coerce_to_optional.submit(allow_failure(s2))
    d_opt = coerce_to_optional.submit(allow_failure(s3))

    # Now pass clean values to the downstream task
    return task_4.submit(c_opt, d_opt)

if __name__ == "__main__":
    client = Client(n_workers=2, threads_per_worker=2)
    task_runner = DaskTaskRunner(address=client.scheduler.address)
    result = main_flow.with_options(task_runner=task_runner)().result(raise_on_failure=False)
    print(f"Final result of the flow: {result}")

Both patterns avoid a direct data dependency on a failed upstream when calling

task_4

, so

task_4

will run and can use whichever inputs are available. If these still fail in your environment, can you share: - Your Prefect and prefect-dask versions (

pip show prefect prefect-dask

) - Whether you’re on Prefect 3.x or 2.x

There were behavior changes between major versions, and an older combo could explain why

allow_failure

didn’t take effect on inputs.