The right way to have a prefect task that always runs after every other task (at the end of the flow)?

Help
6

no, wrap it in allow_failure() instead - you need to do that for all tasks for which you want to allow failure:

@flow(log_prints=True, name="Cleanup task may not get executed")
def main():
    # 1, 2, 3 can run concurrently
    one = task_1.submit()
    two = task_2.submit()
    three = task_3.submit()
    four = task_4.submit(wait_for=[one, two, three])
    five = task_5.submit(allow_failure(four),)
    clean_up_task.submit(wait_for=[allow_failure(four), allow_failure(five)])


if __name__ == "__main__":
    main()

in v2, there are Automations instead: https://docs.prefect.io/ui/automations/