From 72e97fd48e181b1f1bbe65545780db47201a2445 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 29 Sep 2026 10:34:40 -0700 Subject: [PATCH 1/4] Shared one time-skipping unlock between concurrent result waiters. The test server holds one lock per in-flight task and counts one unlock per request, so a second waiter let the clock skip while another workflow still had a task in flight. --- changelog/fixed/shared-sundial.md | 1 + temporalio/testing/_workflow.py | 43 +++++++++++++++++++++-------- tests/testing/test_workflow.py | 45 +++++++++++++++++++++++++++++++ 3 files changed, 78 insertions(+), 11 deletions(-) create mode 100644 changelog/fixed/shared-sundial.md diff --git a/changelog/fixed/shared-sundial.md b/changelog/fixed/shared-sundial.md new file mode 100644 index 000000000..a2466f903 --- /dev/null +++ b/changelog/fixed/shared-sundial.md @@ -0,0 +1 @@ +`WorkflowEnvironment.start_time_skipping()`: concurrent `WorkflowHandle.result()` waiters now share one time-skipping unlock, so the clock no longer jumps while another Workflow still has a task in flight. diff --git a/temporalio/testing/_workflow.py b/temporalio/testing/_workflow.py index 6c8354162..71fbfa2ad 100644 --- a/temporalio/testing/_workflow.py +++ b/temporalio/testing/_workflow.py @@ -690,6 +690,7 @@ def __init__( server: temporalio.bridge.testing.EphemeralServer, ) -> None: self._auto_time_skipping = True + self._time_skipping_waiters = 0 super().__init__( _client_with_interceptors( client, @@ -803,22 +804,42 @@ async def time_skipping_unlocked(self) -> AsyncIterator[None]: if not self._auto_time_skipping: yield None return - await self._client.test_service.unlock_time_skipping( - temporalio.api.testservice.v1.UnlockTimeSkippingRequest() - ) + # The test server holds one time-skipping lock per in-flight task and + # counts one unlock per request, and it skips as soon as the count is + # zero. Concurrent waiters therefore share a single unlock; a second + # one would let the clock jump while another workflow still has a task + # in flight. The waiter is counted before the await so a second waiter + # cannot slip in and unlock again. + self._time_skipping_waiters += 1 + if self._time_skipping_waiters == 1: + try: + await self._client.test_service.unlock_time_skipping( + temporalio.api.testservice.v1.UnlockTimeSkippingRequest() + ) + except BaseException: + self._time_skipping_waiters -= 1 + raise try: yield None - await self._client.test_service.lock_time_skipping( - temporalio.api.testservice.v1.LockTimeSkippingRequest() - ) - except: + except BaseException: + # Lock it back, swallowing error try: - await self._client.test_service.lock_time_skipping( - temporalio.api.testservice.v1.LockTimeSkippingRequest() - ) - except: + await self._lock_time_skipping_after_waiter() + except Exception: logger.exception("Failed locking time skipping after error") raise + # Lock it back, throwing on error + await self._lock_time_skipping_after_waiter() + + async def _lock_time_skipping_after_waiter(self) -> None: + # Only the last waiter locks. The count drops first so a failed lock + # call does not leave a phantom waiter behind. + self._time_skipping_waiters -= 1 + if self._time_skipping_waiters: + return + await self._client.test_service.lock_time_skipping( + temporalio.api.testservice.v1.LockTimeSkippingRequest() + ) class _V2WorkflowEnvironment(_HasAServer): diff --git a/tests/testing/test_workflow.py b/tests/testing/test_workflow.py index a648c4e51..f561a7bb9 100644 --- a/tests/testing/test_workflow.py +++ b/tests/testing/test_workflow.py @@ -176,6 +176,51 @@ async def test_workflow_env_time_skipping_disabled(): assert monotonic() - start > 2.5 +@activity.defn +async def echo(value: str) -> str: + return value + + +@workflow.defn +class TwoActivitiesWorkflow: + @workflow.run + async def run(self, value: str) -> str: + for _ in range(2): + value = await workflow.execute_activity( + echo, value, start_to_close_timeout=timedelta(seconds=15) + ) + return value + + +async def test_workflow_env_time_skipping_concurrent_result_waiters(): + # The test server holds one time-skipping lock per in-flight task and counts + # one unlock per request, so concurrent result() waiters have to share one + # unlock. Otherwise the clock skips as soon as the first workflow completes + # and times out the task the other workflow still has in flight. + async with await WorkflowEnvironment.start_time_skipping() as env: + async with new_worker( + env.client, TwoActivitiesWorkflow, activities=[echo] + ) as worker: + handles = [ + await env.client.start_workflow( + TwoActivitiesWorkflow.run, + value, + id=f"workflow-{uuid.uuid4()}", + task_queue=worker.task_queue, + ) + for value in ("first", "second") + ] + assert ["first", "second"] == await asyncio.gather( + *(handle.result() for handle in handles) + ) + for handle in handles: + async for event in handle.fetch_history_events(): + assert not event.HasField("workflow_task_timed_out_event_attributes") + assert not event.HasField( + "workflow_execution_timed_out_event_attributes" + ) + + @workflow.defn class AssertFailWorkflow: @workflow.run From 665a958cb129fd7d2d2e83117d8f0964c39f0ad5 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 29 Sep 2026 16:34:25 -0700 Subject: [PATCH 2/4] Waited for the target workflow to close before the update that must fail. The last update's result does not mean the target workflow has closed. An update that lands before the closure stays pending until pytest's timeout instead of failing, which hung test_temporal_operation_update_workflow on two Linux lanes. Ported from temporalio/sdk-python#1810. --- tests/nexus/test_temporal_operation.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/tests/nexus/test_temporal_operation.py b/tests/nexus/test_temporal_operation.py index f23728d17..1fa7e0c3e 100644 --- a/tests/nexus/test_temporal_operation.py +++ b/tests/nexus/test_temporal_operation.py @@ -658,6 +658,10 @@ async def run_update(i: int) -> str: async def check_updates_on_completed_workflows_fail(): """The handler workflow already finished at this point, further updaes should just fail""" + # The last update's result does not mean the target has closed yet. + # An update that lands before the closure stays pending instead of + # failing. + await target_handle.result() wf_handle = await client.start_workflow( UpdateWorkflowCaller.run, Input( From 39ec7929f54fbf47759136a94f2d928825c24832 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Thu, 22 Oct 2026 11:55:41 -0700 Subject: [PATCH 3/4] Moved the Nexus update test fix to its own PR. It is unrelated to the time-skipping unlock, so it gets its own review. --- tests/nexus/test_temporal_operation.py | 4 ---- 1 file changed, 4 deletions(-) diff --git a/tests/nexus/test_temporal_operation.py b/tests/nexus/test_temporal_operation.py index 1fa7e0c3e..f23728d17 100644 --- a/tests/nexus/test_temporal_operation.py +++ b/tests/nexus/test_temporal_operation.py @@ -658,10 +658,6 @@ async def run_update(i: int) -> str: async def check_updates_on_completed_workflows_fail(): """The handler workflow already finished at this point, further updaes should just fail""" - # The last update's result does not mean the target has closed yet. - # An update that lands before the closure stays pending instead of - # failing. - await target_handle.result() wf_handle = await client.start_workflow( UpdateWorkflowCaller.run, Input( From 189d1187bec35d24f17aeadb6091df875eff442a Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Thu, 22 Oct 2026 11:56:26 -0700 Subject: [PATCH 4/4] Said why concurrent waiters share one time-skipping unlock. --- temporalio/testing/_workflow.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/temporalio/testing/_workflow.py b/temporalio/testing/_workflow.py index 71fbfa2ad..3d65f84bf 100644 --- a/temporalio/testing/_workflow.py +++ b/temporalio/testing/_workflow.py @@ -804,12 +804,12 @@ async def time_skipping_unlocked(self) -> AsyncIterator[None]: if not self._auto_time_skipping: yield None return - # The test server holds one time-skipping lock per in-flight task and - # counts one unlock per request, and it skips as soon as the count is - # zero. Concurrent waiters therefore share a single unlock; a second - # one would let the clock jump while another workflow still has a task - # in flight. The waiter is counted before the await so a second waiter - # cannot slip in and unlock again. + # Time moves only when nothing is in flight. The server counts one lock per + # in-flight task and one unlock per request, and it skips once the count is zero. + # So concurrent waiters share one unlock. A second unlock would let time jump + # past a task that another Workflow still runs, and that task would time out. + # The waiter is counted before the await, so a second waiter can't unlock again. + # Only V1 needs this. V2 keeps a separate clock for each Workflow. self._time_skipping_waiters += 1 if self._time_skipping_waiters == 1: try: