From 4649730df19f99a779d62bbbd041b3b24c0c1adb Mon Sep 17 00:00:00 2001 From: DABH Date: Thu, 10 Sep 2026 00:52:20 -0500 Subject: [PATCH 1/3] 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 42ba0b69c..4d98fc056 100644 --- a/tests/worker/test_workflow.py +++ b/tests/worker/test_workflow.py @@ -946,7 +946,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 ], @@ -955,7 +955,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 ], @@ -976,16 +976,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( @@ -1003,10 +1006,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( @@ -1022,10 +1028,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( @@ -1042,7 +1051,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 9db8b6cb3235a70625e9a4541fa8cd73ca57124b Mon Sep 17 00:00:00 2001 From: DABH Date: Thu, 10 Sep 2026 00:37:19 -0500 Subject: [PATCH 2/3] 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 cc5be34860bda34ca549206f266314d35e4fef3a Mon Sep 17 00:00:00 2001 From: DABH Date: Thu, 10 Sep 2026 01:37:04 -0500 Subject: [PATCH 3/3] 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 4d98fc056..824a39631 100644 --- a/tests/worker/test_workflow.py +++ b/tests/worker/test_workflow.py @@ -981,6 +981,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, @@ -988,7 +994,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( @@ -1006,13 +1011,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( @@ -1028,13 +1030,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( @@ -1051,9 +1050,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