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 = (