From 6d7c33ab7b9b5420552947fb3131f9af6823f733 Mon Sep 17 00:00:00 2001 From: DABH Date: Tue, 13 Oct 2026 12:54:13 -0500 Subject: [PATCH] Fix workflow test retries and local activity shutdown Keep deliberate warning-as-error retries inside the history assertion deadline, cancel and await their background result task, and retain 60-second task timeouts for the normal workflow tests that require initialization headroom. Use a unique queue for each termination case so a prior server-side poll cannot take its initial task before the intended worker starts. Preserve the coalesced activation checks and all existing assertions. Advance SDK Core to merged 9cb90f5a so queued local activity cancellations reach the language worker before eviction. Import the included Core release notes. --- .../breaking-changes/runtgenoscopic-janyce.md | 1 + changelog/fixed/runtgenoscopic-janyce.md | 2 ++ temporalio/bridge/sdk-core | 2 +- tests/worker/test_workflow.py | 35 ++++++++++++++----- 4 files changed, 31 insertions(+), 9 deletions(-) create mode 100644 changelog/breaking-changes/runtgenoscopic-janyce.md create mode 100644 changelog/fixed/runtgenoscopic-janyce.md diff --git a/changelog/breaking-changes/runtgenoscopic-janyce.md b/changelog/breaking-changes/runtgenoscopic-janyce.md new file mode 100644 index 0000000000..c13dc8f714 --- /dev/null +++ b/changelog/breaking-changes/runtgenoscopic-janyce.md @@ -0,0 +1 @@ +Core: Replay detects changes to the service or operation of scheduled Nexus operations, reporting nondeterminism while retaining compatibility with histories recorded before these checks. diff --git a/changelog/fixed/runtgenoscopic-janyce.md b/changelog/fixed/runtgenoscopic-janyce.md new file mode 100644 index 0000000000..73e2df8151 --- /dev/null +++ b/changelog/fixed/runtgenoscopic-janyce.md @@ -0,0 +1,2 @@ +Core: Local activities still running when their workflow is evicted or the worker shuts down now always receive a cancellation. Local activities no longer retry ordinary failures or start-to-close timeouts after cancellation has been requested, matching non-local activities. Workflow code no longer observes a spurious local activity cancellation when its run is evicted or completes while the local activity is backing off between retries. +Core: Workers no longer panic when evicting a workflow with buffered local activity results. Completed activations' output is recorded before eviction, without recording eviction-induced cancellations. diff --git a/temporalio/bridge/sdk-core b/temporalio/bridge/sdk-core index 472f94eef7..9cb90f5a92 160000 --- a/temporalio/bridge/sdk-core +++ b/temporalio/bridge/sdk-core @@ -1 +1 @@ -Subproject commit 472f94eef798a79cb3424f02f85ae03160791d34 +Subproject commit 9cb90f5a92c75167fe690a39c1ffd91f71a0fdfd diff --git a/tests/worker/test_workflow.py b/tests/worker/test_workflow.py index 01a2fcea89..60461af58e 100644 --- a/tests/worker/test_workflow.py +++ b/tests/worker/test_workflow.py @@ -3884,6 +3884,8 @@ async def test_workflow_patch_activation_callback(client: Client): args=["my-patch", False], id=workflow_id, task_queue=worker.task_queue, + # A timed-out attempt would consult the callback again. + task_timeout=timedelta(seconds=60), ) assert result == [True, True] @@ -3904,6 +3906,8 @@ async def test_workflow_patch_activation_callback_can_decline(client: Client): args=["my-patch", False], id=f"workflow-{uuid.uuid4()}", task_queue=worker.task_queue, + # A timed-out attempt would consult the callback again. + task_timeout=timedelta(seconds=60), ) assert await handle.result() == [False, False] @@ -3939,6 +3943,8 @@ async def test_workflow_patch_activation_callback_not_recalled_on_replay( args=["my-patch", True], id=f"workflow-{uuid.uuid4()}", task_queue=worker.task_queue, + # A timed-out attempt would consult the callback again. + task_timeout=timedelta(seconds=60), ) assert result == [False, False] @@ -6832,6 +6838,8 @@ async def test_workflow_current_update(client: Client): CurrentUpdateWorkflow.run, id=f"wf-{uuid.uuid4()}", task_queue=worker.task_queue, + # Repeated task timeouts would reject updates before their handlers run. + task_timeout=timedelta(seconds=60), ) update_ids = await asyncio.gather( handle.execute_update(CurrentUpdateWorkflow.do_update, id="update1"), @@ -6945,16 +6953,23 @@ async def test_unfinished_handlers_cause_exceptions_in_test_suite(self): # If we don't capture warnings then -- since the unfinished handler warning is converted to # an exception in the test suite -- we see WFT failures when we don't wait for handlers. handle: asyncio.Future[WorkflowHandle] = asyncio.Future() - asyncio.create_task( + result_task = asyncio.create_task( self._get_workflow_result( - wait_all_handlers_finished=False, handle_future=handle + wait_all_handlers_finished=False, + handle_future=handle, + # Keep deliberate task failure/retry inside the history assertion's deadline. + task_timeout=timedelta(seconds=1), ) ) - await assert_eq_eventually( - True, - partial(self._workflow_task_failed, workflow_id=(await handle).id), - timeout=timedelta(seconds=20), - ) + try: + await assert_eq_eventually( + True, + partial(self._workflow_task_failed, workflow_id=(await handle).id), + timeout=timedelta(seconds=20), + ) + finally: + result_task.cancel() + await asyncio.gather(result_task, return_exceptions=True) async def _workflow_task_failed(self, workflow_id: str) -> bool: resp = await self.client.workflow_service.get_workflow_execution_history( @@ -6991,12 +7006,14 @@ async def _get_workflow_result( wait_all_handlers_finished: bool, unfinished_policy: workflow.HandlerUnfinishedPolicy | None = None, handle_future: asyncio.Future[WorkflowHandle] | None = None, + task_timeout: timedelta = timedelta(seconds=60), ) -> bool: handle = await self.client.start_workflow( UnfinishedHandlersWarningsWorkflow.run, arg=wait_all_handlers_finished, id=f"wf-{uuid.uuid4()}", task_queue=self.worker.task_queue, + task_timeout=task_timeout, ) if handle_future: handle_future.set_result(handle) @@ -7179,7 +7196,7 @@ async def test_warning_is_issued_on_exit_with_unfinished_handler( async def _run_workflow_and_get_warning(self) -> bool: workflow_id = f"wf-{uuid.uuid4()}" update_id = "update-id" - task_queue = "tq" + task_queue = str(uuid.uuid4()) # We require a startWorkflow, an update, and maybe a cancellation request, to be delivered # in the same WFT. To do this we start the worker after they've all been accepted by the @@ -7194,6 +7211,8 @@ async def _run_workflow_and_get_warning(self) -> bool: ], id=workflow_id, task_queue=task_queue, + # Repeated task timeouts would reject updates before their handlers run. + task_timeout=timedelta(seconds=60), ) if self.handler_type == "-update-": update_method = (