[Feature Request] Timeout in Workflow definition not in execution

[cosminpm]

Is your feature request related to a problem? Please describe.

I’m trying to set a timeout for a Workflow, but I’m unsure if I’m missing documentation or if this might be a feature request.

I know it’s possible to set a timeout when executing a workflow directly, like this:

result = await client.execute_workflow(
    YourWorkflow.run,
    "your timeout argument",
    id="your-workflow-id",
    task_queue="your-task-queue",
    # Set Workflow Timeout duration
    execution_timeout=timedelta(seconds=2),
    # run_timeout=timedelta(seconds=2),
    # task_timeout=timedelta(seconds=2),
)

However, I’m running into an issue: I can’t set execution_timeout when the workflow is triggered by a Schedule, as my workflows are executed by a worker:

worker = Worker(
    self.client,
    task_queue=TASK_QUEUE,
    workflows=[
        Workflow1,
        Workflow2,
        Workflow3,
    ],
    activities=[
        activity_1,
        activity_2,
        activity_3,
    ],
)

I’ve reviewed the documentation and examples, but haven’t found a way to configure a workflow timeout outside of the direct execution method.

Describe the solution you'd like

Or the workflow has a timeout parameter in the definition or the worker, both of them would solve my issue.

Additional context

N/A

[cretz]

This is a server side timeout set when the workflow is started, and that start is done by the internal scheduler on schedules. How are you creating the schedule? The workflow schedule action allows execution_timeout, so that's where you'd set this. Also, unless you're sure you need it, you should consider using timers inside your workflow that you can react to for timing things out instead of the server-side timeout that terminates a workflow with no recourse.

[cosminpm]

Hello @cretz!

Thanks for answering.

We are creating the schedule via de UI by passing the Workflow definition which does not provide an execution timeout as far as I can see from the available parameters.

I'm implementing workflow timeouts due to an issue where, occasionally, our workflow remains in a "Running" status indefinitely, with zero pending activities. This occurs despite having timeouts set at the activity level, and I haven’t yet identified the root cause.

This is how I'm defining the workflows:

with workflow.unsafe.imports_passed_through():
    from app.module import activity_1


@workflow.defn
class Workflow1:
    @workflow.run
    async def run(self) -> Any:
        return await workflow.execute_activity(
            activity_1_activity,
            start_to_close_timeout=timedelta(minutes=5),
        )


@activity.defn
async def activity_1_activity() -> Any:
    return activity_1()

Am I missing something?

[cretz]

We are creating the schedule via de UI

This seems to be an issue with the UI not providing every workflow option and not related to Python SDK. I will confer with the UI team to see if we have plans here. In the meantime, you could consider creating the schedule via the CLI or programmatically in the Python SDK to get the extra options you need.

I'm implementing workflow timeouts due to an issue where, occasionally, our workflow remains in a "Running" status indefinitely, with zero pending activities. This occurs despite having timeouts set at the activity level, and I haven’t yet identified the root cause.

This is worth identifying instead of just setting workflow execution timeouts. Your activity timeout is only set for each attempt (as opposed to schedule to close timeout), so a failing activity will retry indefinitely there any look like it's hanging (see the UI to see if the pending activity is performing many attempts).

Also, for general assistance with activity timeouts and such, feel free to visit our forums or #python-sdk on Slack.

[cosminpm]

Hello @cretz.

Thanks for answering! Regarding the UI team, do I have to open a request there (if so can you provide a link) or do we keep this issue open?

you could consider creating the schedule via the CLI or programmatically in the Python SDK to get the extra options you need.

Maybe I'm wrong but as far as I know creating a Workflow does not provide the option to set a timeout or at least in the documentation that's not specified, the only references are at Activity level.

The other option for Workflow timeouts are in the Failure detection section, but I think this (got from the official docs) would not work:

# ...
    result = await client.execute_workflow(
        YourWorkflow.run,
        "your timeout argument",
        id="your-workflow-id",
        task_queue="your-task-queue",
        # Set Workflow Timeout duration
        execution_timeout=timedelta(seconds=2),
        # run_timeout=timedelta(seconds=2),
        # task_timeout=timedelta(seconds=2),
    )

As it forces to execute the workflow in your code and my workflows are executed via a schedule. And in the schedule documentation there are no references about workflow timeouts.

This is worth identifying instead of just setting workflow execution timeouts. Your activity timeout is only set for each attempt (as opposed to schedule to close timeout), so a failing activity will retry indefinitely there any look like it's hanging (see the UI to see if the pending activity is performing many attempts).

Indeed, but it's a very strange issue, as sometimes the workflow executes correctly but from now and then it just loops in Running without any pending activity, if we terminate the workflow and we wait for the next workflow execution it succeeds. So it's a non-deterministic behavior. For this reason as a fast solution would be nice to have a Workflow timeout with Schedules.

For sure I may missed something and it's a mistake from my side, but it would be nice if you could help me!

Thanks!

[cretz]

And in the schedule documentation there are no references about workflow timeouts.

The schedule documentation doesn't mention every workflow option since many are the same as regular workflow start. The documentation at https://docs.temporal.io/develop/python/schedules#create shows an example of creating a schedule, you can add workflow options to ScheduleActionStartWorkflow as needed, including execution_timeout (see API documentation at https://python.temporal.io/temporalio.client.ScheduleActionStartWorkflow.html).

[cosminpm]

Thanks @cretz.

Some final questions:

1. Do you have an ETA for when the timeout schedule parameter will be implemented in the UI? If not, can you provide a ticket to follow up on this?
2. I understand that timeouts can be specified via the CLI at the schedule level, but is there a way to define them at the Workflow level without using this option:

    result = await client.execute_workflow(
        YourWorkflow.run,
        "your timeout argument",
        id="your-workflow-id",
        task_queue="your-task-queue",
        # Set Workflow Timeout duration
        execution_timeout=timedelta(seconds=2),
        # run_timeout=timedelta(seconds=2),
        # task_timeout=timedelta(seconds=2),
    )

Thanks!

[cretz]

Do you have an ETA for when the timeout schedule parameter will be implemented in the UI? If not, can you provide a ticket to follow up on this?

I do not have an ETA and there may be discussions on use cases the UI team would want to have with you and confirm whether it's something they want to expose. Instead of in the Python GitHub issue, I would recommend starting a UI discussion in a more general setting, e.g. at https://community.temporal.io/ or in Slack (https://t.mp/slack) or, if you're a cloud customer, open a ticket.

I understand that timeouts can be specified via the CLI at the schedule level, but is there a way to define them at the Workflow level without using this option:

Can you clarify the question a bit? The execution_timeout option is available both for create schedule in Python (when you use the workflow action I mentioned above) and for just starting a workflow directly in Python. What is the issue with using the execution_timeout option?

[cosminpm]

Yes for sure, I have requested it in the #web-ui channel. Thanks!

2. Well, my point if we could have something like this:

from datetime import timedelta
from typing import Any
from temporalio import workflow, activity

with workflow.unsafe.imports_passed_through():
    from app.module import activity_1


@workflow.defn
class Workflow1:
    def __init__(self, timeout: timedelta = timedelta(minutes=5)): # <- HERE
        self.timeout = timeout

    @workflow.run
    async def run(self) -> Any:
        return await workflow.execute_activity(
            activity_1_activity,
            start_to_close_timeout=timedelta(minutes=5),
        )


@activity.defn
async def activity_1_activity() -> Any:
    return activity_1()

When the scheduler executes the workflow, it knows what is the timeout for each one, and this timeout it's defined at Workflow level, this would also simplify this code as you won't need to specify it in the client.

    result = await client.execute_workflow(
        YourWorkflow.run,
        "your timeout argument",
        id="your-workflow-id",
        task_queue="your-task-queue",
        # Set Workflow Timeout duration
        execution_timeout=timedelta(seconds=2), # <- HERE
        # run_timeout=timedelta(seconds=2),
        # task_timeout=timedelta(seconds=2),
    )

[cretz]

Still not following exactly. If you want timeout inside workflow code, you can have (untested):

@dataclass
class Workflow1Args:
    timeout: int

@workflow.defn
class Workflow1:
    async def run(self, args: Workflow1Args) -> None:
        try:
            async with asyncio.timeout(args.timeout):
                # This will get canceled on timeout
                my_other_code
        except TimeoutError: # Note, may be something else depending on how cancel handled, make sure to test
            raise ApplicationError('timeout')

We recommend this approach over execution_timeout because execution_timeout is server side and you can't react to it. But if you must have server side timeout (e.g. you may not have a worker running), then no amount of in-workflow timeout will help because the workflow code may never be reached.

If you just want to create a schedule with an execution timeout, just:

    await client.create_schedule(
        "workflow-schedule-id",
        Schedule(
            action=ScheduleActionStartWorkflow(
                YourSchedulesWorkflow.run,
                "my schedule arg",
                id="schedules-workflow-id",
                task_queue="schedules-task-queue",
                execution_timeout=timedelta(seconds=2)
            ),
            spec=ScheduleSpec(
                intervals=[ScheduleIntervalSpec(every=timedelta(minutes=2))]
            ),
            state=ScheduleState(note="Here's a note on my Schedule."),
        ),
    )

(though a 2-second timeout may be a bit aggressive)

[cosminpm]

Thanks! I think this solves my question, now I will be following on the UI problem.

[cosminpm]

Follow up issue: temporalio/ui#2460