Hi, I'm working on my first flow and trying to wrap my head around tasks states and the trigger of "any failed". For reference, I've got a very simple flow like this:

with Flow('test flow') as f:
	data = get_data()
	cleansed_data = munge_data(data)
	send_data = email_data(cleansed_data)
	alert = send_error_email()

	alert.set_upstream(send_data)

f.set_reference_tasks([send_data])

I want my

alert

task to trigger if any of the tasks above fail. So, in my task definition I have this set:

@task(trigger = prefect.triggers.any_failed)
def send_error_email():

When the flow runs the first three tasks complete successfully, however and alert task fails, because the trigger fails. (TRIGGERFAIL signal raised during execution.) However, the overall Flow status is a success because I've set the flow's reference task to the

send_data

task Is this the correct way to use the trigger of "any_failed". It seems to me that it might be more natural for the alert task to be in status of "skipped" if none of the upstream tasks failed.

Hi Mark - great question; ultimately yes you are using the trigger correctly. Honestly I’ll have to get back to you on why we fail (instead of skip) based on triggers; I’m sure there was a reason but I’ll need to think on it and get back to you!

thanks, Chris. I appreciate the confirmation!

@Chris White Might this be the reason? From the docs:

SKIP is treated like SUCCESS
When a task is skipped, it's usually treated as if it ran successfully. This is because tasks only skip if users specifically introduce skipping logic, so the result is compliant with the user's design.

So as far as I understand, the

SKIP

signal is reserved to be raised exclusively by the user.