Hi, I'd like to write a task that is triggered on triggers.always_run, but take two different actions depending on if any tasks failed, or if all tasks were successful. Any pointers? As in something similar to:

@task(trigger=prefect.triggers.always_run)  # one new line
def tear_down_cluster(cluster):
    if all_successful:
        print(":)")
    else:
        print(":-/")

Hey @Philip MacMenamin, I think the best way to approach this is to add some downstream dependent tasks that take actions based on the final state of upstream tasks, rather than attach an

always_run

trigger to a task with conditional logic. Here’s a great example in the documentation of taking two courses of action based on the final state of

task_a

, giving you granular visibility across the final states of your tasks and the logic engaged.

Hey Kyle - ok, I noticed that. The reason I was looking at the always_run was I wasn't 100% sure how to guarantee that a task is going to run, regardless of other failures, and always_run seemed to be the way to do this. What would a way to do this be setting one task to trigger on any_failed, and another to trigger on all_succesdful. Naturally I don't want to ever enter a case where both of these tasks are fired, it needs to only ever be one.

Hmm, in that case you may be better off with a Flow Level State Handler that fires based on the final state of a flow. If you configure your tasks to bubble their failure to the Flow Level, then you can send one of those two notifications based on the final state of the flow. You can also set a State Handler notification at the task level as needed, that looks to the final state of a task(s) instead.

Haha, so this is what I started out with believe it or not. But there was a problem with all my states being None. I'll try it again... I might have messed something up.

Interesting… Nice job! 🚀