feat(workflow): support workflow failure

Signed-off-by: Shijie Sheng <[email protected]>

Author: shijiesheng

cadence/_internal/workflow/workflow_engine.py

@@ -1,21 +1,24 @@
 import logging
 from dataclasses import dataclass
+import traceback
 from typing import List
 
 from cadence._internal.workflow.context import Context
 from cadence._internal.workflow.decision_events_iterator import DecisionEventsIterator
 from cadence._internal.workflow.statemachine.decision_manager import DecisionManager
 from cadence._internal.workflow.workflow_intance import WorkflowInstance
-from cadence.api.v1.common_pb2 import Payload
+from cadence.api.v1.common_pb2 import Failure, Payload
 from cadence.api.v1.decision_pb2 import (
     CompleteWorkflowExecutionDecisionAttributes,
     Decision,
+    FailWorkflowExecutionDecisionAttributes,
 )
 from cadence.api.v1.history_pb2 import (
     HistoryEvent,
     WorkflowExecutionStartedEventAttributes,
 )
 from cadence.api.v1.service_worker_pb2 import PollForDecisionTaskResponse
+from cadence.error import WorkflowFailure
 from cadence.workflow import WorkflowDefinition, WorkflowInfo
 
 logger = logging.getLogger(__name__)
@@ -75,23 +78,32 @@ def process_decision(
                 decisions = self._decision_manager.collect_pending_decisions()
 
                 # complete workflow if it is done
-                try:
-                    if self._workflow_instance.is_done():
+                if self._workflow_instance.is_done():
+                    try:
                         result = self._workflow_instance.get_result()
+                    except WorkflowFailure as e:
+                        decisions.append(
+                            Decision(
+                                fail_workflow_execution_decision_attributes=FailWorkflowExecutionDecisionAttributes(
+                                    failure=Failure(
+                                        reason=str(e),
+                                        details=traceback.format_exc().encode("utf-8"),
+                                    )
+                                )
+                            )
+                        )
+                    # TODO: handle cancellation error
+                    except Exception:
+                        raise
+                    else:
                         decisions.append(
                             Decision(
                                 complete_workflow_execution_decision_attributes=CompleteWorkflowExecutionDecisionAttributes(
                                     result=result
                                 )
                             )
                         )
-                    return DecisionResult(decisions=decisions)
-
-                except Exception:
-                    # TODO: handle CancellationError
-                    # TODO: handle WorkflowError
-                    # TODO: handle unknown error, fail decision task and try again instead of breaking the engine
-                    raise
+                return DecisionResult(decisions=decisions)
 
         except Exception as e:
             # Log decision task failure with full context (matches Java ReplayDecisionTaskHandler)

cadence/_internal/workflow/workflow_intance.py

@@ -1,8 +1,9 @@
-from asyncio import Task
+from asyncio import CancelledError, InvalidStateError, Task
 from typing import Any, Optional
 from cadence._internal.workflow.deterministic_event_loop import DeterministicEventLoop
 from cadence.api.v1.common_pb2 import Payload
 from cadence.data_converter import DataConverter
+from cadence.error import WorkflowFailure
 from cadence.workflow import WorkflowDefinition
 
 
@@ -33,6 +34,11 @@ def is_done(self) -> bool:
     def get_result(self) -> Payload:
         if self._task is None:
             raise RuntimeError("Workflow is not started yet")
-        result = self._task.result()
+        try:
+            result = self._task.result()
+        except (CancelledError, InvalidStateError) as e:
+            raise e
+        except Exception as e:
+            raise WorkflowFailure(f"Workflow failed: {e}") from e
         # TODO: handle result with multiple outputs
         return self._data_converter.to_data([result])

cadence/error.py

@@ -6,6 +6,11 @@ def __init__(self, message: str) -> None:
         super().__init__(message)
 
 
+class WorkflowFailure(Exception):
+    def __init__(self, message: str) -> None:
+        super().__init__(message)
+
+
 class CadenceRpcError(Exception):
     def __init__(self, message: str, code: grpc.StatusCode, *args):
         super().__init__(message, code, *args)

tests/integration_tests/workflow/test_workflows.py

@@ -23,6 +23,13 @@ async def echo(self, message: str) -> str:
         return message
 
 
+@reg.workflow()
+class FailureWorkflow:
+    @workflow.run
+    async def failure(self, message: str) -> str:
+        raise Exception("mocked workflow failure")
+
+
 async def test_simple_workflow(helper: CadenceHelper):
     async with helper.worker(reg) as worker:
         execution = await worker.client.start_workflow(
@@ -50,6 +57,40 @@ async def test_simple_workflow(helper: CadenceHelper):
         )
 
 
+async def test_workflow_failure(helper: CadenceHelper):
+    async with helper.worker(reg) as worker:
+        execution = await worker.client.start_workflow(
+            "FailureWorkflow",
+            "hello world",
+            task_list=worker.task_list,
+            execution_start_to_close_timeout=timedelta(seconds=10),
+        )
+
+        response: GetWorkflowExecutionHistoryResponse = await worker.client.workflow_stub.GetWorkflowExecutionHistory(
+            GetWorkflowExecutionHistoryRequest(
+                domain=DOMAIN_NAME,
+                workflow_execution=execution,
+                wait_for_new_event=True,
+                history_event_filter_type=EventFilterType.EVENT_FILTER_TYPE_CLOSE_EVENT,
+                skip_archival=True,
+            )
+        )
+
+        assert (
+            "Workflow failed: mocked workflow failure"
+            == response.history.events[
+                -1
+            ].workflow_execution_failed_event_attributes.failure.reason
+        )
+
+        assert (
+            """raise Exception("mocked workflow failure")"""
+            in response.history.events[
+                -1
+            ].workflow_execution_failed_event_attributes.failure.details.decode()
+        )
+
+
 @pytest.mark.skip(reason="Incorrect WorkflowType")
 async def test_workflow_fn(helper: CadenceHelper):
     async with helper.worker(reg) as worker: