From c49f6642b482d5aa02c44be08f1ef125ced83fe3 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Sun, 6 Sep 2026 07:52:00 +0000 Subject: [PATCH 1/2] Reproduce missing terminal workflow timeout result --- src/durable_workflow/__init__.py | 2 + src/durable_workflow/errors.py | 11 +++++ tests/test_workflow_result_timeout.py | 61 +++++++++++++++++++++++++++ 3 files changed, 74 insertions(+) create mode 100644 tests/test_workflow_result_timeout.py diff --git a/src/durable_workflow/__init__.py b/src/durable_workflow/__init__.py index f1dd4d3..399c14e 100644 --- a/src/durable_workflow/__init__.py +++ b/src/durable_workflow/__init__.py @@ -95,6 +95,7 @@ WorkflowNotFound, WorkflowPayloadDecodeError, WorkflowTerminated, + WorkflowTimedOut, ) from .external_storage import ( EXTERNAL_PAYLOAD_REFERENCE_SCHEMA, @@ -374,6 +375,7 @@ "WorkflowFailed", "WorkflowNotFound", "WorkflowTerminated", + "WorkflowTimedOut", "EXTERNAL_TASK_INPUT_CONTRACT_SCHEMA", "EXTERNAL_TASK_INPUT_MEDIA_TYPE", "EXTERNAL_TASK_INPUT_SCHEMA", diff --git a/src/durable_workflow/errors.py b/src/durable_workflow/errors.py index 4120cc3..613448e 100644 --- a/src/durable_workflow/errors.py +++ b/src/durable_workflow/errors.py @@ -560,6 +560,17 @@ def __init__(self, message: str = "workflow was terminated") -> None: super().__init__(message) +class WorkflowTimedOut(DurableWorkflowError): + """A persisted workflow execution or run deadline expired. + + Unlike a caller's :class:`TimeoutError` while polling, this is a terminal + workflow outcome recorded by the runtime. + """ + + def __init__(self, message: str = "workflow execution timed out") -> None: + super().__init__(message) + + class SagaCompensationFailed(DurableWorkflowError): """A saga compensation failed after an earlier workflow failure. diff --git a/tests/test_workflow_result_timeout.py b/tests/test_workflow_result_timeout.py new file mode 100644 index 0000000..2f98d47 --- /dev/null +++ b/tests/test_workflow_result_timeout.py @@ -0,0 +1,61 @@ +from unittest.mock import AsyncMock + +import pytest + +from durable_workflow import Client, WorkflowTimedOut +from durable_workflow.client import WorkflowExecution, WorkflowHandle +from durable_workflow.errors import WorkflowCancelled, WorkflowFailed, WorkflowTerminated + + +def result_client(status: str, event_type: str, payload: dict) -> tuple[Client, WorkflowHandle]: + client = Client("https://unused.example") + client.describe_workflow = AsyncMock( + return_value=WorkflowExecution( + workflow_id="order", run_id="current-run", workflow_type="order", status=status, + ) + ) + client.get_history = AsyncMock( + return_value={"events": [{"event_type": event_type, "payload": payload}]} + ) + return client, WorkflowHandle(client, workflow_id="order", run_id="selected-run", workflow_type="order") + + +@pytest.mark.parametrize("kind", ["execution_timeout", "run_timeout"]) +async def test_persisted_deadline_raises_typed_timeout_for_selected_history(kind: str) -> None: + client, handle = result_client( + "failed", "WorkflowTimedOut", {"timeout_kind": kind, "deadline_at": "2026-01-01T00:00:00Z"}, + ) + try: + with pytest.raises(WorkflowTimedOut, match="workflow execution timed out"): + await client.get_result(handle, timeout=0) + client.get_history.assert_awaited_once_with("order", "selected-run") + finally: + await client.aclose() + + +async def test_polling_timeout_is_not_a_terminal_workflow_timeout() -> None: + client, handle = result_client("waiting", "ConditionWaitOpened", {}) + try: + with pytest.raises(TimeoutError, match="not terminal") as caught: + await client.get_result(handle, timeout=0) + assert not isinstance(caught.value, WorkflowTimedOut) + client.get_history.assert_not_awaited() + finally: + await client.aclose() + + +@pytest.mark.parametrize( + ("status", "event_type", "exception"), + [ + ("failed", "WorkflowFailed", WorkflowFailed), + ("cancelled", "WorkflowCancelled", WorkflowCancelled), + ("terminated", "WorkflowTerminated", WorkflowTerminated), + ], +) +async def test_other_terminal_failures_preserve_their_exception(status, event_type, exception) -> None: + client, handle = result_client(status, event_type, {"message": "stopped", "reason": "stopped"}) + try: + with pytest.raises(exception, match="stopped"): + await client.get_result(handle, timeout=0) + finally: + await client.aclose() From 3a9ad63435cec799b5e068320d494609a9c57c6f Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Sun, 6 Sep 2026 07:53:35 +0000 Subject: [PATCH 2/2] Raise WorkflowTimedOut for persisted runtime deadlines --- src/durable_workflow/client.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/src/durable_workflow/client.py b/src/durable_workflow/client.py index ad7a4f9..715e3b2 100644 --- a/src/durable_workflow/client.py +++ b/src/durable_workflow/client.py @@ -46,6 +46,7 @@ WorkflowCancelled, WorkflowFailed, WorkflowTerminated, + WorkflowTimedOut, _raise_for_status, ) from .external_storage import ( @@ -4141,8 +4142,9 @@ async def get_result( Raises :class:`~durable_workflow.errors.WorkflowFailed`, :class:`~durable_workflow.errors.WorkflowCancelled`, or :class:`~durable_workflow.errors.WorkflowTerminated` if the workflow - ended in a non-success state, or :class:`TimeoutError` if ``timeout`` - seconds elapse before the workflow terminates. + ended in a non-success state. A persisted execution or run deadline + raises :class:`~durable_workflow.errors.WorkflowTimedOut`; a caller's + :class:`TimeoutError` means ``timeout`` seconds elapsed while waiting. """ deadline = asyncio.get_running_loop().time() + timeout while True: @@ -4177,6 +4179,8 @@ async def get_result( raise WorkflowCancelled( payload.get("reason", "workflow was cancelled") ) + if etype == "WorkflowTimedOut": + raise WorkflowTimedOut() return None if asyncio.get_running_loop().time() > deadline: raise TimeoutError(