diff --git a/CHANGELOG.md b/CHANGELOG.md index 483c73d1d..fd2c78b91 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -82,6 +82,8 @@ to include examples, links to docs, or any other relevant information. ### Fixed +- Continue-as-new requests from workflow update handlers now fail the workflow + task instead of leaving the update unresolved. - Marked system Nexus envelope payloads so nested payloads can be detected and visited after the envelope is already stored as a payload. diff --git a/temporalio/worker/_workflow_instance.py b/temporalio/worker/_workflow_instance.py index d0b10ccae..9d956aac4 100644 --- a/temporalio/worker/_workflow_instance.py +++ b/temporalio/worker/_workflow_instance.py @@ -761,6 +761,14 @@ async def run_update() -> None: else: self._current_activation_error = err return + except _ContinueAsNewError: + # Continue-as-new is only valid from the top-level workflow. + # From an update handler it must fail the workflow task + # instead of escaping this detached task. + self._current_activation_error = RuntimeError( + "Cannot continue as new from an update handler" + ) + return except BaseException: if self._deleting: if LOG_IGNORE_DURING_DELETE: diff --git a/tests/worker/test_workflow.py b/tests/worker/test_workflow.py index 283357c2d..68861bada 100644 --- a/tests/worker/test_workflow.py +++ b/tests/worker/test_workflow.py @@ -5259,6 +5259,7 @@ async def do_stuff(buffer: MetricBuffer) -> None: bad_validator_fail_ct = 0 task_fail_ct = 0 +continue_as_new_update_task_fail_ct = 0 @workflow.defn @@ -5342,6 +5343,14 @@ def throws_runtime_err(self) -> None: task_fail_ct += 1 raise RuntimeError("intentional failure") + @workflow.update + def continue_as_new_for_task_failure(self) -> str: + global continue_as_new_update_task_fail_ct + if continue_as_new_update_task_fail_ct < 1: + continue_as_new_update_task_fail_ct += 1 + workflow.continue_as_new() + return "recovered after task failure" + async def test_workflow_update_handlers_happy(client: Client): async with new_worker( @@ -5484,6 +5493,31 @@ async def test_workflow_update_task_fails(client: Client): assert bad_validator_fail_ct == 2 +async def test_workflow_update_continue_as_new_fails_task(client: Client): + # Continue-as-new is a control-flow BaseException. From an update handler + # it must fail the workflow task, then retry the update rather than letting + # the detached update task escape without completing the protocol message. + global continue_as_new_update_task_fail_ct + continue_as_new_update_task_fail_ct = 0 + async with new_worker( + client, UpdateHandlersWorkflow, workflow_runner=UnsandboxedWorkflowRunner() + ) as worker: + handle = await client.start_workflow( + UpdateHandlersWorkflow.run, + id=f"update-continue-as-new-{uuid.uuid4()}", + task_queue=worker.task_queue, + task_timeout=timedelta(seconds=1), + ) + result = await asyncio.wait_for( + handle.execute_update( + UpdateHandlersWorkflow.continue_as_new_for_task_failure + ), + timeout=timedelta(seconds=10).total_seconds(), + ) + assert result == "recovered after task failure" + assert continue_as_new_update_task_fail_ct == 1 + + @workflow.defn class UpdateRespectsFirstExecutionRunIdWorkflow: def __init__(self) -> None: