From 085e368d3427aca622591af5bbb9f0d8f9016667 Mon Sep 17 00:00:00 2001
From: DABH <DABH@users.noreply.github.com>
Date: Thu, 10 Sep 2026 00:30:16 -0500
Subject: [PATCH] Fix activity pause test races

The pause tests raced their own activities:

- unpause_and_assert required the activity to still be pending after the
  unpause RPC, but the retried attempt sees heartbeat details and completes
  immediately, so a slow describe observed no pending activity.
- The pause workflows set heartbeat timeouts of 1-2s while relying on the
  default throttle (80% of the timeout), leaving a few hundred milliseconds
  before the server timed the attempt out and the pause RPC or the next
  describe found no pending activity. Pause is delivered through heartbeat
  responses whose cadence is the worker throttle interval, so drop the
  timeouts and set a 300ms throttle explicitly.
- assert_pending_activity_exists_eventually waited only 5s for the first
  workflow task to schedule the activity, which a cold worker on a loaded
  runner exceeds; use the 10s the other helpers use.

Also assert that the external heartbeat raises after the pause instead of
silently passing when it does not.
---
 tests/helpers/__init__.py     | 13 ++++++++-----
 tests/worker/test_workflow.py | 16 ++++++----------
 2 files changed, 14 insertions(+), 15 deletions(-)

diff --git a/tests/helpers/__init__.py b/tests/helpers/__init__.py
index fe37296e9..242bac1e6 100644
--- a/tests/helpers/__init__.py
+++ b/tests/helpers/__init__.py
@@ -239,7 +239,7 @@ async def check_workflow_exists() -> bool:
 async def assert_pending_activity_exists_eventually(
     handle: WorkflowHandle,
     activity_id: str,
-    timeout: timedelta = timedelta(seconds=5),
+    timeout: timedelta = timedelta(seconds=10),
 ) -> PendingActivityInfo:
     """Wait until a pending activity with the given ID exists and return it."""
 
@@ -351,7 +351,11 @@ async def check_paused() -> None:
 
 
 async def unpause_and_assert(client: Client, handle: WorkflowHandle, activity_id: str):
-    """Unpause the given activity and assert it is not paused."""
+    """Unpause the given activity and assert it is no longer paused.
+
+    An unpaused activity may retry and close before we observe it, so an
+    activity that is no longer pending also counts as unpaused.
+    """
     desc = await handle.describe()
     req = UnpauseActivityRequest(
         namespace=client.namespace,
@@ -363,10 +367,9 @@ async def unpause_and_assert(client: Client, handle: WorkflowHandle, activity_id
     )
     await client.workflow_service.unpause_activity(req)
 
-    # Assert eventually not paused
     async def check_unpaused() -> None:
-        info = await assert_pending_activity_exists_eventually(handle, activity_id)
-        assert not info.paused, f"Activity {activity_id} still paused"
+        info = await get_pending_activity_info(handle, activity_id)
+        assert info is None or not info.paused, f"Activity {activity_id} still paused"
 
     await assert_eventually(check_unpaused)
 
diff --git a/tests/worker/test_workflow.py b/tests/worker/test_workflow.py
index 42ba0b69c..2466fa338 100644
--- a/tests/worker/test_workflow.py
+++ b/tests/worker/test_workflow.py
@@ -8938,7 +8938,6 @@ async def run(
                 True,
                 activity_id=activity_id,
                 start_to_close_timeout=timedelta(seconds=10),
-                heartbeat_timeout=timedelta(seconds=2),
                 retry_policy=RetryPolicy(maximum_attempts=1),
             )
         )
@@ -8948,7 +8947,6 @@ async def run(
                 True,
                 activity_id=f"{activity_id}-2",
                 start_to_close_timeout=timedelta(seconds=10),
-                heartbeat_timeout=timedelta(seconds=2),
                 retry_policy=RetryPolicy(maximum_attempts=1),
             )
         )
@@ -8967,6 +8965,8 @@ async def test_activity_pause_cancellation_details(
             workflows=[ActivityHeartbeatWorkflow],
             activities=[heartbeat_activity, sync_heartbeat_activity],
             activity_executor=executor,
+            max_heartbeat_throttle_interval=timedelta(milliseconds=300),
+            default_heartbeat_throttle_interval=timedelta(milliseconds=300),
         ) as worker:
             test_activity_id = f"heartbeat-activity-{uuid.uuid4()}"
 
@@ -9019,7 +9019,6 @@ async def run(
                 False,
                 activity_id=activity_id,
                 start_to_close_timeout=timedelta(seconds=10),
-                heartbeat_timeout=timedelta(seconds=1),
                 retry_policy=RetryPolicy(maximum_attempts=2),
             )
         )
@@ -9029,7 +9028,6 @@ async def run(
                 False,
                 activity_id=f"{activity_id}-2",
                 start_to_close_timeout=timedelta(seconds=10),
-                heartbeat_timeout=timedelta(seconds=1),
                 retry_policy=RetryPolicy(maximum_attempts=2),
             )
         )
@@ -9127,7 +9125,6 @@ async def run(self, activity_id: str) -> None:
             external_activity_heartbeat,
             activity_id=activity_id,
             start_to_close_timeout=timedelta(seconds=10),
-            heartbeat_timeout=timedelta(seconds=1),
             retry_policy=RetryPolicy(maximum_attempts=2),
         )
 
@@ -9167,12 +9164,11 @@ async def test_external_activity_cancellation_details(
         # Pause activity then assert it is paused
         await pause_and_assert(client, wf_handle, activity_info.activity_id)
 
-        try:
+        with pytest.raises(AsyncActivityCancelledError) as err:
             await external_activity_handle.heartbeat()
-        except AsyncActivityCancelledError as err:
-            assert err.details == temporalio.activity.ActivityCancellationDetails(
-                paused=True
-            )
+        assert err.value.details == temporalio.activity.ActivityCancellationDetails(
+            paused=True
+        )
 
 
 @activity.defn