Great question. In Prefect 3.x you can let a task fail without failing the flow (and without retries) by: - Not awaiting the failing task’s result in the flow - If a downstream task depends on it, wrap the upstream future with

allow_failure(...)

so the dependency doesn’t cancel, then handle the failure inside the downstream task Here’s a minimal example:

from prefect import flow, task, allow_failure, get_run_logger

@task
def might_fail(n: int) -> int:
    logger = get_run_logger()
    if n % 2:
        <http://logger.info|logger.info>("About to fail")
        raise ValueError("boom")
    return n * 2

@task
def consume_optional_input(upstream):
    # `upstream` comes from `allow_failure(...)`
    # Safely get the result without raising if it failed
    val_or_exc = upstream.result(raise_on_failure=False)
    if isinstance(val_or_exc, Exception):
        # upstream task is marked Failed; handle it and continue
        get_run_logger().warning(f"Upstream failed: {val_or_exc}")
        return "fallback"
    else:
        return f"got {val_or_exc}"

@flow
def my_flow():
    a = might_fail.submit(1)          # This will fail and be marked Failed
    b = might_fail.submit(2)          # This will succeed

    # This downstream depends on `a`, but will still run even if `a` failed
    c = consume_optional_input.submit(allow_failure(a))

    # You can keep running other tasks; avoid calling a.result() with default args
    # so you don’t raise in the flow
    _ = b.result()  # fine to resolve successes

    # Flow will complete normally because we didn't raise at the flow level
    return {"c": c.result(raise_on_failure=False)}

Notes: - Prefect will only retry if you set retries; by default there are no retries, so you’re already covered on the “do not retry” requirement. - If you don’t need a downstream dependency on the failing task at all, the simplest option is to just not depend on it and don’t call

.result()

on its future in the flow. The task will be marked Failed, but the flow will still finish. - When you do depend on a task that may fail, wrap the future with

allow_failure(...)

. In the downstream task, use

.result(raise_on_failure=False)

to read the value (or get the exception) and handle it without raising. Related references: - Prefect 3.x release notes (mentions availability of

allow_failure

) If you can share a small snippet of your current flow, I can tailor this pattern to match your exact dependency graph.