Fix activity pause test races

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)
 

tests/worker/test_workflow.py

@@ -8941,7 +8941,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),
             )
         )
@@ -8951,7 +8950,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),
             )
         )
@@ -8970,6 +8968,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()}"
 
@@ -9022,7 +9022,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),
             )
         )
@@ -9032,7 +9031,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),
             )
         )
@@ -9130,7 +9128,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),
         )
 
@@ -9170,12 +9167,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