Hi everyone, I found a weird behaviour when I use a timeout with a LocalDaskExecutor and manually raise a Success. The task will always fail with the timeout error message, even if it takes less time than the specified timeout. Minimal example:

from prefect import Flow, task
from prefect.executors import LocalDaskExecutor

from prefect.engine.signals import SUCCESS


@task(timeout=60)
def succeed():
    raise SUCCESS()


with Flow("timeout-test") as flow:
    succeed()

flow.executor = LocalDaskExecutor()
flow.register(project_name="tutorial")

Is this the intended behaviour? Can anyone reproduce it?

Hi @Bastian Röhrig, thanks for the minimal example here! I can reproduce. I don't believe this is the intended behavior, I'll check with the team.