Hello Prefect team, a question about retrying tasks conditionally. In our task state handler we look at the failure message every time a task fails and we want to retry on some of those errors. Per this doc entry: https://docs.prefect.io/core/concepts/execution.html#state-signals we perform a

raise signals.RETRY

in those cases. Problem is this does not appear to work in the task handler. This code gets called: https://github.com/PrefectHQ/prefect/blob/5b7015a75ea60bc5a8ffb676b3be2ff406103583/src/prefect/engine/cloud/task_runner.py#L96-L105 and the task ultimately is failed, not retried. So should our code be something more like this:

return Retrying(msg=msg, context=new_state.context)

or should this in fact work? Thanks!

Hey @Jeff Brainerd! Great question, let me take a look

Thanks for getting back @Dylan! I will keep my eye on this.

Of course!