Diverge

Author: dandavison

message_passing/waiting_for_handlers/README.md

@@ -1,12 +1,15 @@
-# Waiting for message handlers, and performing compensation and cleanup in message handlers
+# Waiting for message handlers in message handlers
 
-This sample demonstrates how to do the following:
+This workflow demonstrates how to wait for signal and update handlers to
+finish in the following circumstances:
 
-1. Ensure that all update/signal handlers are finished before a successful
-    workflow return, and on workflow cancellation and failure.
-2. Perform compensation/cleanup in an update handler when the workflow is
-    cancelled or fails.
+- Before a successful return
+- On failure
+- On cancellation
 
+Your workflow can also exit via Continue-As-New. In that case you would
+usually wait for the handlers to finish immediately before the call to
+continue_as_new(); that's not illustrated in this sample.
 
 
 To run, open two terminals and `cd` to this directory in them.
@@ -29,11 +32,10 @@ workflow exit type: SUCCESS
 
 
 workflow exit type: FAILURE
-    🔴 caught exception while waiting for update result: Workflow update failed: The update failed because the workflow run exited
+    🟢 caller received update result
     🔴 caught exception while waiting for workflow result: Workflow execution failed: deliberately failing workflow
 
 
 workflow exit type: CANCELLATION
-    🔴 caught exception while waiting for update result: Workflow update failed: The update failed because the workflow run exited
-    🔴 caught exception while waiting for workflow result: Workflow execution failed: Workflow cancelled
+    🟢 caller received update result
 ```

message_passing/waiting_for_handlers/__init__.py

@@ -14,3 +14,8 @@ class WorkflowExitType(IntEnum):
 @dataclass
 class WorkflowInput:
     exit_type: WorkflowExitType
+
+
+@dataclass
+class WorkflowResult:
+    data: str

message_passing/waiting_for_handlers/activities.py

@@ -3,16 +3,6 @@
 from temporalio import activity
 
 
-@activity.defn
-async def activity_executed_to_perform_workflow_compensation():
-    await asyncio.sleep(1)
-
-
 @activity.defn
 async def activity_executed_by_update_handler():
     await asyncio.sleep(1)
-
-
-@activity.defn
-async def activity_executed_by_update_handler_to_perform_compensation():
-    await asyncio.sleep(1)

message_passing/waiting_for_handlers/starter.py

@@ -2,21 +2,21 @@
 
 from temporalio import client, common
 
-from message_passing.waiting_for_handlers_and_compensation import (
+from message_passing.waiting_for_handlers import (
     TASK_QUEUE,
     WORKFLOW_ID,
     WorkflowExitType,
     WorkflowInput,
 )
-from message_passing.waiting_for_handlers_and_compensation.workflows import (
-    WaitingForHandlersAndCompensationWorkflow,
+from message_passing.waiting_for_handlers.workflows import (
+    WaitingForHandlersWorkflow,
 )
 
 
 async def starter(exit_type: WorkflowExitType):
     cl = await client.Client.connect("localhost:7233")
     wf_handle = await cl.start_workflow(
-        WaitingForHandlersAndCompensationWorkflow.run,
+        WaitingForHandlersWorkflow.run,
         WorkflowInput(exit_type=exit_type),
         id=WORKFLOW_ID,
         task_queue=TASK_QUEUE,
@@ -31,11 +31,13 @@ async def _check_run(
 ):
     try:
         up_handle = await wf_handle.start_update(
-            WaitingForHandlersAndCompensationWorkflow.my_update,
+            WaitingForHandlersWorkflow.my_update,
             wait_for_stage=client.WorkflowUpdateStage.ACCEPTED,
         )
     except Exception as e:
-        print(f"    🔴 caught exception while starting update: {e}: {e.__cause__ or ''}")
+        print(
+            f"    🔴 caught exception while starting update: {e}: {e.__cause__ or ''}"
+        )
 
     if exit_type == WorkflowExitType.CANCELLATION:
         await wf_handle.cancel()
@@ -51,7 +53,7 @@ async def _check_run(
     try:
         await wf_handle.result()
         print("    🟢 caller received workflow result")
-    except Exception as e:
+    except BaseException as e:
         print(
             f"    🔴 caught exception while waiting for workflow result: {e}: {e.__cause__ or ''}"
         )

message_passing/waiting_for_handlers/worker.py

@@ -4,14 +4,12 @@
 from temporalio.client import Client
 from temporalio.worker import Worker
 
-from message_passing.waiting_for_handlers_and_compensation import TASK_QUEUE
-from message_passing.waiting_for_handlers_and_compensation.activities import (
+from message_passing.waiting_for_handlers import TASK_QUEUE
+from message_passing.waiting_for_handlers.activities import (
     activity_executed_by_update_handler,
-    activity_executed_by_update_handler_to_perform_compensation,
-    activity_executed_to_perform_workflow_compensation,
 )
-from message_passing.waiting_for_handlers_and_compensation.workflows import (
-    WaitingForHandlersAndCompensationWorkflow,
+from message_passing.waiting_for_handlers.workflows import (
+    WaitingForHandlersWorkflow,
 )
 
 interrupt_event = asyncio.Event()
@@ -25,11 +23,9 @@ async def main():
     async with Worker(
         client,
         task_queue=TASK_QUEUE,
-        workflows=[WaitingForHandlersAndCompensationWorkflow],
+        workflows=[WaitingForHandlersWorkflow],
         activities=[
             activity_executed_by_update_handler,
-            activity_executed_by_update_handler_to_perform_compensation,
-            activity_executed_to_perform_workflow_compensation,
         ],
     ):
         logging.info("Worker started, ctrl+c to exit")

message_passing/waiting_for_handlers/workflows.py

@@ -1,180 +1,95 @@
 import asyncio
 from datetime import timedelta
-from typing import cast
 
 from temporalio import exceptions, workflow
 
-from message_passing.waiting_for_handlers_and_compensation import (
+from message_passing.waiting_for_handlers import (
     WorkflowExitType,
     WorkflowInput,
+    WorkflowResult,
 )
-from message_passing.waiting_for_handlers_and_compensation.activities import (
+from message_passing.waiting_for_handlers.activities import (
     activity_executed_by_update_handler,
-    activity_executed_by_update_handler_to_perform_compensation,
-    activity_executed_to_perform_workflow_compensation,
 )
 
 
-@workflow.defn
-class WaitingForHandlersAndCompensationWorkflow:
+def is_workflow_exit_exception(e: BaseException) -> bool:
     """
-    This Workflow demonstrates how to wait for message handlers to finish and
-    perform compensation/cleanup:
+    True if the exception is of a type that will cause the workflow to exit.
 
-    1. It ensures that all signal and update handlers have finished before a
-       successful return, and on failure and cancellation.
-    2. The update handler performs any necessary compensation/cleanup when the
-       workflow is cancelled or fails.
+    This is as opposed to exceptions that cause a workflow task failure, which
+    are retried automatically by Temporal.
     """
+    # 👉 If you have set additional failure_exception_types you should also
+    # check for these here.
+    return isinstance(e, (asyncio.CancelledError, exceptions.FailureError))
 
-    def __init__(self) -> None:
-        # 👉 If the workflow exits prematurely, this future will be completed
-        # with the associated exception as its value. Message handlers can then
-        # "race" this future against a task performing the message handler's own
-        # application logic; if this future completes before the message handler
-        # task then the handler should abort and perform compensation.
-        self.workflow_exit: asyncio.Future[None] = asyncio.Future()
-
-        # The following two attributes are implementation detail of this sample
-        # and can be ignored
-        self._update_started = False
-        self._update_compensation_done = False
-        self._workflow_compensation_done = False
 
+@workflow.defn
+class WaitingForHandlersWorkflow:
     @workflow.run
-    async def run(self, input: WorkflowInput) -> str:
+    async def run(self, input: WorkflowInput) -> WorkflowResult:
+        """
+        This workflow.run method demonstrates a pattern that can be used to wait for signal and
+        update handlers to finish in the following circumstances:
+
+        - On successful workflow return
+        - On workflow cancellation
+        - On workflow failure
+
+        Your workflow can also exit via Continue-As-New. In that case you would usually wait for
+        the handlers to finish immediately before the call to continue_as_new(); that's not
+        illustrated in this sample.
+
+        If you additionally need to perform cleanup or compensation on workflow failure or
+        cancellation, see the message_passing/waiting_for_handlers_and_compensation sample.
+        """
         try:
             # 👉 Use this `try...except` style, instead of waiting for message
             # handlers to finish in a `finally` block. The reason is that some
             # exception types cause a workflow task failure as opposed to
             # workflow exit, in which case we do *not* want to wait for message
             # handlers to finish.
-
-            # 👉 self._run contains your actual application logic. This is
-            # implemented in a separate method in order to separate
-            # "platform-level" concerns (waiting for handlers to finish and
-            # ensuring that compensation is performed when appropriate) from
-            # application logic. In this sample, its actual implementation is
-            # below but contains nothing relevant.
-            result = await self._run(input)
-            self.workflow_exit.set_result(None)
+            result = await self._my_workflow_application_logic(input)
             await workflow.wait_condition(workflow.all_handlers_finished)
             return result
         # 👉 Catch BaseException since asyncio.CancelledError does not inherit
         # from Exception.
         except BaseException as e:
             if is_workflow_exit_exception(e):
-                self.workflow_exit.set_exception(e)
                 await workflow.wait_condition(workflow.all_handlers_finished)
-                await self.workflow_compensation()
-                self._workflow_compensation_done = True
             raise
 
-    async def workflow_compensation(self):
-        await workflow.execute_activity(
-            activity_executed_to_perform_workflow_compensation,
-            start_to_close_timeout=timedelta(seconds=10),
-        )
-        self._update_compensation_done = True
+    # Methods below this point can be ignored unless you are interested in
+    # the implementation details of this sample.
+
+    def __init__(self) -> None:
+        self._update_started = False
 
     @workflow.update
     async def my_update(self) -> str:
-        """
-        An update handler that handles exceptions raised in its own execution
-        and in that of the main workflow method.
-
-        It ensures that:
-        - Compensation/cleanup is always performed when appropriate
-        - The update caller gets the update result, or WorkflowUpdateFailedError
-        """
-        # 👉 As with the main workflow method, the update application logic is
-        # implemented in a separate method in order to separate "platform-level"
-        # error-handling and compensation concerns from application logic. Note
-        # that coroutines must be wrapped in tasks in order to use
-        # workflow.wait.
-        update_task = asyncio.create_task(self._my_update())
-
-        # 👉 "Race" the workflow_exit future against the handler's own application
-        # logic. Always use `workflow.wait` instead of `asyncio.wait` in
-        # Workflow code: asyncio's version is non-deterministic.
-        await workflow.wait(  # type: ignore
-            [update_task, self.workflow_exit], return_when=asyncio.FIRST_EXCEPTION
-        )
-        try:
-            if update_task.done():
-                # 👉 The update has finished (whether successfully or not).
-                # Regardless of whether the main workflow method is about to
-                # exit or not, the update caller should receive a response
-                # informing them of the outcome of the update. So return the
-                # result, or raise the exception that caused the update handler
-                # to exit.
-                return await update_task
-            else:
-                # 👉 The main workflow method exited prematurely due to an
-                # error, and this happened before the update finished. Fail the
-                # update with the workflow exception as cause.
-                raise exceptions.ApplicationError(
-                    "The update failed because the workflow run exited"
-                ) from cast(BaseException, self.workflow_exit.exception())
-        # 👉 Catch BaseException since asyncio.CancelledError does not inherit
-        # from Exception.
-        except BaseException as e:
-            if is_workflow_exit_exception(e):
-                try:
-                    await self.my_update_compensation()
-                except BaseException as e:
-                    raise exceptions.ApplicationError(
-                        "Update compensation failed"
-                    ) from e
-            raise
-
-    async def my_update_compensation(self):
-        await workflow.execute_activity(
-            activity_executed_by_update_handler_to_perform_compensation,
-            start_to_close_timeout=timedelta(seconds=10),
-        )
-        self._update_compensation_done = True
-
-    @workflow.query
-    def workflow_compensation_done(self) -> bool:
-        return self._workflow_compensation_done
-
-    @workflow.query
-    def update_compensation_done(self) -> bool:
-        return self._update_compensation_done
-
-    # The following methods are placeholders for the actual application logic
-    # that you would perform in your main workflow method  or update handler.
-    # Their implementation can be ignored.
-
-    async def _my_update(self) -> str:
-        # Ignore this method unless you are interested in the implementation
-        # details of this sample.
         self._update_started = True
         await workflow.execute_activity(
             activity_executed_by_update_handler,
             start_to_close_timeout=timedelta(seconds=10),
         )
         return "update-result"
 
-    async def _run(self, input: WorkflowInput) -> str:
-        # Ignore this method unless you are interested in the implementation
-        # details of this sample.
+    async def _my_workflow_application_logic(
+        self, input: WorkflowInput
+    ) -> WorkflowResult:
+        # The main workflow logic is implemented in a separate method in order
+        # to separate "platform-level" concerns (waiting for handlers to finish
+        # and error handling) from application logic.
 
         # Wait until handlers have started, so that we are demonstrating that we
         # wait for them to finish.
         await workflow.wait_condition(lambda: self._update_started)
         if input.exit_type == WorkflowExitType.SUCCESS:
-            return "workflow-result"
+            return WorkflowResult(data="workflow-result")
         elif input.exit_type == WorkflowExitType.FAILURE:
             raise exceptions.ApplicationError("deliberately failing workflow")
         elif input.exit_type == WorkflowExitType.CANCELLATION:
             # Block forever; the starter will send a workflow cancellation request.
             await asyncio.Future()
         raise AssertionError("unreachable")
-
-
-def is_workflow_exit_exception(e: BaseException) -> bool:
-    # 👉 If you have set additional failure_exception_types you should also
-    # check for these here.
-    return isinstance(e, (asyncio.CancelledError, exceptions.FailureError))