fix: report workflow and update compensation independently

message_passing/waiting_for_handlers_and_compensation/workflows.py

@@ -77,7 +77,6 @@ async def workflow_compensation(self):
             activity_executed_to_perform_workflow_compensation,
             start_to_close_timeout=timedelta(seconds=10),
         )
-        self._update_compensation_done = True
 
     @workflow.update
     async def my_update(self) -> str:

tests/message_passing/waiting_for_handlers_and_compensation/waiting_for_handlers_and_compensation_test.py

@@ -2,7 +2,8 @@
 from enum import Enum
 
 import pytest
-from temporalio import client, worker
+from temporalio import activity, client, exceptions, worker
+from temporalio.api.enums.v1 import EventType
 from temporalio.testing import WorkflowEnvironment
 
 from message_passing.waiting_for_handlers_and_compensation import (
@@ -104,3 +105,122 @@ async def test_waiting_for_handlers_and_compensation(
             assert await wf_handle.query(
                 WaitingForHandlersAndCompensationWorkflow.workflow_compensation_done
             )
+
+        await assert_compensation_state(
+            wf_handle,
+            workflow_compensated=workflow_expect == WorkflowExpect.FAILURE,
+            update_compensated=update_expect == UpdateExpect.FAILURE,
+        )
+
+
+async def test_workflow_compensation_without_update(env: WorkflowEnvironment):
+    async with worker.Worker(
+        env.client,
+        task_queue=TASK_QUEUE,
+        workflows=[WaitingForHandlersAndCompensationWorkflow],
+        activities=[
+            activity_executed_by_update_handler,
+            activity_executed_by_update_handler_to_perform_compensation,
+            activity_executed_to_perform_workflow_compensation,
+        ],
+    ):
+        wf_handle = await env.client.start_workflow(
+            WaitingForHandlersAndCompensationWorkflow.run,
+            WorkflowInput(exit_type=WorkflowExitType.CANCELLATION),
+            id=str(uuid.uuid4()),
+            task_queue=TASK_QUEUE,
+        )
+        # Ensure the workflow is running before cancelling without an update.
+        assert not await wf_handle.query(
+            WaitingForHandlersAndCompensationWorkflow.workflow_compensation_done
+        )
+        await wf_handle.cancel()
+        with pytest.raises(client.WorkflowFailureError) as err:
+            await wf_handle.result()
+        assert isinstance(err.value.cause, exceptions.CancelledError)
+
+        await assert_compensation_state(
+            wf_handle, workflow_compensated=True, update_compensated=False
+        )
+
+
+async def test_update_compensation_without_workflow_compensation(
+    env: WorkflowEnvironment,
+):
+    if env.supports_time_skipping:
+        pytest.skip(
+            "Java test server: https://github.com/temporalio/sdk-java/issues/1903"
+        )
+
+    @activity.defn(name="activity_executed_by_update_handler")
+    async def failing_update_activity():
+        raise exceptions.ApplicationError(
+            "deliberately failing update activity", non_retryable=True
+        )
+
+    async with worker.Worker(
+        env.client,
+        task_queue=TASK_QUEUE,
+        workflows=[WaitingForHandlersAndCompensationWorkflow],
+        activities=[
+            failing_update_activity,
+            activity_executed_by_update_handler_to_perform_compensation,
+            activity_executed_to_perform_workflow_compensation,
+        ],
+    ):
+        wf_handle = await env.client.start_workflow(
+            WaitingForHandlersAndCompensationWorkflow.run,
+            WorkflowInput(exit_type=WorkflowExitType.SUCCESS),
+            id=str(uuid.uuid4()),
+            task_queue=TASK_QUEUE,
+        )
+        with pytest.raises(client.WorkflowUpdateFailedError) as err:
+            await wf_handle.execute_update(
+                WaitingForHandlersAndCompensationWorkflow.my_update
+            )
+        assert isinstance(err.value.cause, exceptions.ActivityError)
+        assert (await wf_handle.result()).data == "workflow-result"
+
+        await assert_compensation_state(
+            wf_handle, workflow_compensated=False, update_compensated=True
+        )
+
+
+async def assert_compensation_state(
+    wf_handle: client.WorkflowHandle,
+    *,
+    workflow_compensated: bool,
+    update_compensated: bool,
+):
+    history = await wf_handle.fetch_history()
+    scheduled_activities = {
+        event.event_id: event.activity_task_scheduled_event_attributes.activity_type.name
+        for event in history.events
+        if event.event_type == EventType.EVENT_TYPE_ACTIVITY_TASK_SCHEDULED
+    }
+    completed_activities = [
+        scheduled_activities[
+            event.activity_task_completed_event_attributes.scheduled_event_id
+        ]
+        for event in history.events
+        if event.event_type == EventType.EVENT_TYPE_ACTIVITY_TASK_COMPLETED
+    ]
+    for activity_name, expected in {
+        activity_executed_to_perform_workflow_compensation.__name__: workflow_compensated,
+        activity_executed_by_update_handler_to_perform_compensation.__name__: update_compensated,
+    }.items():
+        assert list(scheduled_activities.values()).count(activity_name) == int(expected)
+        assert completed_activities.count(activity_name) == int(expected)
+
+    assert (
+        await wf_handle.query(
+            WaitingForHandlersAndCompensationWorkflow.workflow_compensation_done
+        )
+        == workflow_compensated
+    )
+    assert (
+        await wf_handle.query(
+            WaitingForHandlersAndCompensationWorkflow.update_compensation_done
+        )
+        == update_compensated
+    )