import time
from prefect import task, flow
from prefect.task_runners import DaskTaskRunner

@task
def this_will_fail(name):
    raise Exception(f"I will fail {name}")

@task
def always_pending():
    print(f"Sleeping for 3 secs")
    time.sleep(3)
    
@task
def normal_task():
    print("doing stuff")
    time.sleep(1)
    
@flow(task_runner=DaskTaskRunner)
def always_pending_flow():
    fail_a = this_will_fail('a')
    fail_b = this_will_fail('b')
    wont_fail = normal_task()
    always_pending(wait_for=[fail_a, wont_fail, fail_b])
    
if __name__ == "__main__": 
    state = always_pending_flow()

ConcurrentTaskRunner

DaskTaskRunner

always_pending