Hi team! Given that theres no support at the moment for flow concurrency in prefect-server, Im trying to make sure that a large influx of late tasks dont swamp our server and bring it down.  Is there a way to automatically cancel tasks that are more than a timedelta late?  For example, if a task is 1 minute late, go ahead and run it. But if its 10 hours late, cancel it.

Hi Samuel, This isn't currently supported by Prefect Server. You could roll your own though by comparing the current time in a task to the scheduled start time in

prefect.context

. The following (untested) decorator might work:

import functools
from datetime import timedelta

import pendulum
from prefect.engine.signals import SKIP


def skip_if_late_by(timedelta):
    def inner(func):
        @functools.wraps(func):
        def wrapped(*args, **kwargs):
            if prefect.context["scheduled_start_time"] + timedelta > pendulum.now():
                raise SKIP("Too late!")
            return func(*args, **kwargs)
        return wrapped
    return inner

@task
@skip_if_late_by(timedelta(hours=10))
def my_task(a, b, c):
    ...

could raise

FAIL

instead to fail it (we should add an easy signal option for

Cancelled

too).

Thanks Jim, Ill give that a shot. What would be the difference in outcome between raising SKIP and FAIL?

Raising

SKIP

will mark the task as skipped (and downstream tasks will usually also skip by default). Raising

FAIL

will mark the task as failed (and downstream tasks will usually not run since the upstream task didn't succeed).

Is there an ETA on The SLA feature? We have a lot of use for a feature like this and would love to try out and offer feedback