Hi, in Prefect v1 I could raise a signal to set a task to a given state like this:

start_time = datetime.utcnow() + timedelta(seconds=retry_after)
        raise signals.RETRY(
            start_time=start_time,
            message=f"Rate limited. Waiting {retry_after}s.",
        )

This way I could implement my own retry logic (e.g. not based only on exception class, but exception data and define

retry_after

based on the API response). Is it possible with Prefect v2? I am searching the docs, but couldn’t find if returning or raising states manually is supported. Something like this:

start_time = pendulum.now("UTC").add(seconds=retry_after)
        return AwaitingRetry(
            scheduled_time=start_time,
            message=f"Rate limited. Waiting {retry_after}s.",
        )

Hi Wojciech! In general, returning a state from a task should work in Prefect 2. In case you're interested, this is the section of code that converts task return values into states. As you can see, if you return a manually-created state, Prefect will use it. It looks like the docs mention this for flows, but not tasks - so I'll make a note to update the docs to make them more clear.

Hi, thanks for the response. I tried returning a state from a task and it seems to work for Completed state, but when returning

AwaitingRetry

like:

start_time = pendulum.now("UTC").add(seconds=retry_after)
        return AwaitingRetry(
            scheduled_time=start_time,
            message=f"Rate limited. Waiting {retry_after}s.",
        )

I am getting:

This run cannot transition to the StateType.SCHEDULED state from the StateType.RUNNING state.

inside

set_task_run_state

. Does it mean we cannot manually retry task with custom

scheduled_time

?