Skip to content
Open
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand Down
8 changes: 8 additions & 0 deletions temporalio/worker/_workflow_instance.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 or
Comment thread
Sakshamm-Goyal marked this conversation as resolved.
Outdated
# signal handlers. 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:
Expand Down
34 changes: 34 additions & 0 deletions tests/worker/test_workflow.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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:
Expand Down
Loading