From da198dfc7065a19e73fa1dbd47c18a2ed805ff38 Mon Sep 17 00:00:00 2001 From: jw_ond Date: Fri, 16 Oct 2026 07:55:45 +0800 Subject: [PATCH] fix: report workflow and update compensation independently --- .../workflows.py | 1 - ...ting_for_handlers_and_compensation_test.py | 122 +++++++++++++++++- 2 files changed, 121 insertions(+), 2 deletions(-) diff --git a/message_passing/waiting_for_handlers_and_compensation/workflows.py b/message_passing/waiting_for_handlers_and_compensation/workflows.py index 450dff2c1..525831019 100644 --- a/message_passing/waiting_for_handlers_and_compensation/workflows.py +++ b/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: diff --git a/tests/message_passing/waiting_for_handlers_and_compensation/waiting_for_handlers_and_compensation_test.py b/tests/message_passing/waiting_for_handlers_and_compensation/waiting_for_handlers_and_compensation_test.py index 2b10d3965..8fbee0e42 100644 --- a/tests/message_passing/waiting_for_handlers_and_compensation/waiting_for_handlers_and_compensation_test.py +++ b/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 + )