From 818f51cf9465f63181a2853fca35cc784bbe363a Mon Sep 17 00:00:00 2001 From: DABH Date: Thu, 10 Sep 2026 00:52:20 -0500 Subject: [PATCH 1/4] Take test_workflow_cancel_activity off one-second clocks The test ran its three cancellation phases with a 1s workflow task timeout, a 1s heartbeat timeout on the regular activity and a 5s schedule-to-close on the local activity. Core only completes the workflow task holding a local activity at 80% of the task timeout, so the completion had 200ms of slack; every failing CI log starts with "Error reporting WFT to server" evictions from missing that. The eviction then either orphaned the local activity (shutdown hang, fixed separately), replayed the task into a stale local activity resolution (Core nondeterminism, queries rejected with "Workflow Task in failed state"), or simply left the phase without its expected query result for 10s. Widen the task timeout to 5s, give the local activity a timeout it never needs to hit, keep cancel delivery fast for the regular activity through the worker heartbeat throttle instead of a 1s heartbeat timeout, and make the shared completion event checks depend only on this phase's activity, with bounded waits so a missing completion fails instead of hanging. --- tests/worker/test_workflow.py | 29 ++++++++++++++++++++--------- 1 file changed, 20 insertions(+), 9 deletions(-) diff --git a/tests/worker/test_workflow.py b/tests/worker/test_workflow.py index d751fdb8b..7db60a19c 100644 --- a/tests/worker/test_workflow.py +++ b/tests/worker/test_workflow.py @@ -947,7 +947,7 @@ async def run(self, params: CancelActivityWorkflowParams) -> None: if params.local: handle = workflow.start_local_activity_method( ActivityWaitCancelNotify.wait_cancel, - schedule_to_close_timeout=timedelta(seconds=5), + schedule_to_close_timeout=timedelta(minutes=1), cancellation_type=workflow.ActivityCancellationType[ params.cancellation_type ], @@ -956,7 +956,7 @@ async def run(self, params: CancelActivityWorkflowParams) -> None: handle = workflow.start_activity_method( ActivityWaitCancelNotify.wait_cancel, schedule_to_close_timeout=timedelta(seconds=5), - heartbeat_timeout=timedelta(seconds=1), + heartbeat_timeout=timedelta(seconds=5), cancellation_type=workflow.ActivityCancellationType[ params.cancellation_type ], @@ -977,16 +977,19 @@ def activity_result(self) -> str: @pytest.mark.parametrize("local", [True, False]) async def test_workflow_cancel_activity(client: Client, local: bool): - # Need short task timeout to timeout LA task and longer assert timeout - # so the task can timeout - task_timeout = timedelta(seconds=1) + # Core completes the task holding a local activity at 80% of this timeout + task_timeout = timedelta(seconds=5) assert_timeout = timedelta(seconds=10) activity_inst = ActivityWaitCancelNotify() async with new_worker( - client, CancelActivityWorkflow, activities=[activity_inst.wait_cancel] + client, + CancelActivityWorkflow, + activities=[activity_inst.wait_cancel], + max_heartbeat_throttle_interval=timedelta(milliseconds=300), ) as worker: # Try cancel - confirm error and activity was sent the cancel + activity_inst.wait_cancel_complete.clear() handle = await client.start_workflow( CancelActivityWorkflow.run, CancelActivityWorkflowParams( @@ -1004,10 +1007,13 @@ async def activity_result() -> str: await assert_eq_eventually( "Error: CancelledError", activity_result, timeout=assert_timeout ) - await activity_inst.wait_cancel_complete.wait() + await asyncio.wait_for( + activity_inst.wait_cancel_complete.wait(), assert_timeout.total_seconds() + ) await handle.cancel() # Wait cancel - confirm no error due to graceful cancel handling + activity_inst.wait_cancel_complete.clear() handle = await client.start_workflow( CancelActivityWorkflow.run, CancelActivityWorkflowParams( @@ -1023,10 +1029,13 @@ async def activity_result() -> str: activity_result, timeout=assert_timeout, ) - await activity_inst.wait_cancel_complete.wait() + await asyncio.wait_for( + activity_inst.wait_cancel_complete.wait(), assert_timeout.total_seconds() + ) await handle.cancel() # Abandon - confirm error and that activity stays running + activity_inst.wait_cancel_complete.clear() handle = await client.start_workflow( CancelActivityWorkflow.run, CancelActivityWorkflowParams( @@ -1043,7 +1052,9 @@ async def activity_result() -> str: await asyncio.sleep(0.5) assert not activity_inst.wait_cancel_complete.is_set() await handle.cancel() - await activity_inst.wait_cancel_complete.wait() + await asyncio.wait_for( + activity_inst.wait_cancel_complete.wait(), assert_timeout.total_seconds() + ) @workflow.defn From d5df5f6b86e0ad07187c83c5c205f07e2fe7700c Mon Sep 17 00:00:00 2001 From: DABH Date: Thu, 10 Sep 2026 00:37:19 -0500 Subject: [PATCH 2/4] Give manual activity tests time for the start handshake test_manual_heartbeat and its siblings start a standalone activity with a 5s start-to-close timeout, then wait for the activity to report that it started through an EventWorkflow before completing, cancelling, failing or heartbeating it by ID. That handshake includes the first workflow task of a cold worker, which on a loaded macOS runner took over 5s; the attempt timed out and the heartbeat failed with RPCError NOT_FOUND ("activity not found for ID"). Nothing in these tests depends on the attempt timing out, so give the attempt a minute. --- tests/test_activity.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/tests/test_activity.py b/tests/test_activity.py index a5682f221..53ffe835e 100644 --- a/tests/test_activity.py +++ b/tests/test_activity.py @@ -752,7 +752,7 @@ async def test_manual_completion(client: Client, env: WorkflowEnvironment): ActivityInput(event_workflow_id=event_workflow_id), id=activity_id, task_queue=task_queue, - start_to_close_timeout=timedelta(seconds=5), + start_to_close_timeout=timedelta(minutes=1), ) async with Worker( @@ -794,7 +794,7 @@ async def test_manual_cancellation(client: Client, env: WorkflowEnvironment): ActivityInput(event_workflow_id=event_workflow_id), id=activity_id, task_queue=task_queue, - start_to_close_timeout=timedelta(seconds=5), + start_to_close_timeout=timedelta(minutes=1), ) async with Worker( @@ -855,7 +855,7 @@ async def test_manual_failure(client: Client, env: WorkflowEnvironment): ActivityInput(event_workflow_id=event_workflow_id), id=activity_id, task_queue=task_queue, - start_to_close_timeout=timedelta(seconds=5), + start_to_close_timeout=timedelta(minutes=1), ) async with Worker( client, @@ -932,7 +932,7 @@ async def test_manual_heartbeat(client: Client, env: WorkflowEnvironment): ), id=activity_id, task_queue=task_queue, - start_to_close_timeout=timedelta(seconds=5), + start_to_close_timeout=timedelta(minutes=1), ) wait_for_activity_start_wf_handle = await client.start_workflow( EventWorkflow.wait, From 63987edf2073df8b8e9247a6e15bb2493d538ecb Mon Sep 17 00:00:00 2001 From: DABH Date: Thu, 10 Sep 2026 01:37:04 -0500 Subject: [PATCH 3/4] Share one cancel-wait helper in test_workflow_cancel_activity --- tests/worker/test_workflow.py | 21 +++++++++------------ 1 file changed, 9 insertions(+), 12 deletions(-) diff --git a/tests/worker/test_workflow.py b/tests/worker/test_workflow.py index 7db60a19c..a7acd9f91 100644 --- a/tests/worker/test_workflow.py +++ b/tests/worker/test_workflow.py @@ -982,6 +982,12 @@ async def test_workflow_cancel_activity(client: Client, local: bool): assert_timeout = timedelta(seconds=10) activity_inst = ActivityWaitCancelNotify() + async def wait_cancel_complete() -> None: + await asyncio.wait_for( + activity_inst.wait_cancel_complete.wait(), assert_timeout.total_seconds() + ) + activity_inst.wait_cancel_complete.clear() + async with new_worker( client, CancelActivityWorkflow, @@ -989,7 +995,6 @@ async def test_workflow_cancel_activity(client: Client, local: bool): max_heartbeat_throttle_interval=timedelta(milliseconds=300), ) as worker: # Try cancel - confirm error and activity was sent the cancel - activity_inst.wait_cancel_complete.clear() handle = await client.start_workflow( CancelActivityWorkflow.run, CancelActivityWorkflowParams( @@ -1007,13 +1012,10 @@ async def activity_result() -> str: await assert_eq_eventually( "Error: CancelledError", activity_result, timeout=assert_timeout ) - await asyncio.wait_for( - activity_inst.wait_cancel_complete.wait(), assert_timeout.total_seconds() - ) + await wait_cancel_complete() await handle.cancel() # Wait cancel - confirm no error due to graceful cancel handling - activity_inst.wait_cancel_complete.clear() handle = await client.start_workflow( CancelActivityWorkflow.run, CancelActivityWorkflowParams( @@ -1029,13 +1031,10 @@ async def activity_result() -> str: activity_result, timeout=assert_timeout, ) - await asyncio.wait_for( - activity_inst.wait_cancel_complete.wait(), assert_timeout.total_seconds() - ) + await wait_cancel_complete() await handle.cancel() # Abandon - confirm error and that activity stays running - activity_inst.wait_cancel_complete.clear() handle = await client.start_workflow( CancelActivityWorkflow.run, CancelActivityWorkflowParams( @@ -1052,9 +1051,7 @@ async def activity_result() -> str: await asyncio.sleep(0.5) assert not activity_inst.wait_cancel_complete.is_set() await handle.cancel() - await asyncio.wait_for( - activity_inst.wait_cancel_complete.wait(), assert_timeout.total_seconds() - ) + await wait_cancel_complete() @workflow.defn From 3df29846a335b21756001f26f12cde87d97dffcb Mon Sep 17 00:00:00 2001 From: DABH Date: Wed, 30 Sep 2026 13:57:34 -0500 Subject: [PATCH 4/4] Give the cancel-activity asserts headroom over two task heartbeats With a 5s task timeout the local-activity phases take two 80% heartbeat cycles, about 8s, before the cancel lands, so a 10s assert budget left under 2s of slack on loaded runners. Use 30s. --- tests/worker/test_workflow.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/tests/worker/test_workflow.py b/tests/worker/test_workflow.py index a7acd9f91..dcb71a231 100644 --- a/tests/worker/test_workflow.py +++ b/tests/worker/test_workflow.py @@ -977,9 +977,11 @@ def activity_result(self) -> str: @pytest.mark.parametrize("local", [True, False]) async def test_workflow_cancel_activity(client: Client, local: bool): - # Core completes the task holding a local activity at 80% of this timeout + # Core completes the task holding a local activity at 80% of this timeout, and + # the cancel reaches the activity on the second such cycle (~8s), so the assert + # budget needs headroom beyond that on loaded runners task_timeout = timedelta(seconds=5) - assert_timeout = timedelta(seconds=10) + assert_timeout = timedelta(seconds=30) activity_inst = ActivityWaitCancelNotify() async def wait_cancel_complete() -> None: