PC-5206: wait out Integration Service, never re-send timed-out activity

packages/uipath-platform/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "uipath-platform"
-version = "0.2.41"
+version = "0.2.42"
 description = "HTTP client library for programmatic access to UiPath Platform"
 readme = { file = "README.md", content-type = "text/markdown" }
 requires-python = ">=3.11"

packages/uipath-platform/src/uipath/platform/connections/_connections_service.py

@@ -3,7 +3,8 @@
 from typing import Any, Dict, List, Optional
 from urllib.parse import parse_qsl, quote, urlsplit
 
-from httpx import Response
+from httpx import ReadTimeout, Response, Timeout
+from tenacity import retry_if_exception, retry_if_result
 from uipath.core.tracing import traced
 
 from ..common._base_service import BaseService
@@ -12,6 +13,7 @@
 from ..common._execution_context import UiPathExecutionContext
 from ..common._folder_context import header_folder
 from ..common._models import Endpoint, RequestSpec
+from ..common.retry import is_retryable_platform_exception, is_retryable_response
 from ..orchestrator._folder_service import FolderService
 from .connections import (
     ActivityMetadata,
@@ -31,6 +33,30 @@
 HEADER_ACTIVITY_JOB_ID = "x-uipath-job-id"
 _ORIGINATOR_VALUE = "uipath-python"
 
+# Integration Service waits up to ~120 s on the vendor behind an activity (a customer's
+# Kusto query, a web read), so the default 30 s client timeout hangs up while it is still
+# working. Wait longer than Integration Service does for the response.
+_ACTIVITY_TIMEOUT = Timeout(30.0, read=150.0)
+
+
+def _is_retryable_activity_exception(exception: BaseException) -> bool:
+    """Like `is_retryable_platform_exception`, but never retry a read timeout.
+
+    A read timeout means Integration Service received the activity and may still run
+    it: sending it again would run a non-idempotent activity twice.
+    """
+    if isinstance(exception, ReadTimeout):
+        return False
+    return is_retryable_platform_exception(exception)
+
+
+_ACTIVITY_RETRY = retry_if_exception(
+    _is_retryable_activity_exception
+) | retry_if_result(is_retryable_response)
+# BaseService.request with its wait and stop strategies, retrying `_ACTIVITY_RETRY` only.
+_request_activity = BaseService.request.retry_with(retry=_ACTIVITY_RETRY)  # type: ignore[attr-defined]
+_request_activity_async = BaseService.request_async.retry_with(retry=_ACTIVITY_RETRY)  # type: ignore[attr-defined]
+
 
 def _to_param_str(value: Any) -> str:
     """Convert an activity parameter value to its wire string; booleans become lowercase."""
@@ -678,6 +704,8 @@ def invoke_activity(
         Raises:
             ValueError: If required parameters are missing or invalid
             RuntimeError: If the HTTP request fails or returns an error status
+            httpx.ReadTimeout: If Integration Service does not answer within 150 s.
+                It is not retried: Integration Service may still be running the activity.
         """
         connection = self.retrieve(connection_id)
         folder_key = connection.folder.get("key") if connection.folder else None
@@ -686,13 +714,15 @@ def invoke_activity(
             activity_metadata, connection.id, activity_input, folder_key
         )
 
-        response = self.request(
+        response = _request_activity(
+            self,
             spec.method,
             url=spec.endpoint,
             headers=spec.headers,
             params=spec.params,
             json=spec.json,
             files=spec.files,
+            timeout=_ACTIVITY_TIMEOUT,
         )
 
         return response.json()
@@ -720,6 +750,8 @@ async def invoke_activity_async(
         Raises:
             ValueError: If required parameters are missing or invalid
             RuntimeError: If the HTTP request fails or returns an error status
+            httpx.ReadTimeout: If Integration Service does not answer within 150 s.
+                It is not retried: Integration Service may still be running the activity.
         """
         connection = await self.retrieve_async(connection_id)
         folder_key = connection.folder.get("key") if connection.folder else None
@@ -728,13 +760,15 @@ async def invoke_activity_async(
             activity_metadata, connection.id, activity_input, folder_key
         )
 
-        response = await self.request_async(
+        response = await _request_activity_async(
+            self,
             spec.method,
             url=spec.endpoint,
             headers=spec.headers,
             params=spec.params,
             json=spec.json,
             files=spec.files,
+            timeout=_ACTIVITY_TIMEOUT,
         )
 
         return response.json()

packages/uipath-platform/tests/services/test_connections_service.py

@@ -2,6 +2,7 @@
 from unittest.mock import AsyncMock, MagicMock
 from urllib.parse import unquote_plus
 
+import httpx
 import pytest
 from pydantic import ValidationError
 from pytest_httpx import HTTPXMock
@@ -1478,6 +1479,123 @@ def test_invoke_activity_omits_job_id_header_when_unset(
         sent_request = httpx_mock.get_requests()[1]
         assert HEADER_ACTIVITY_JOB_ID not in sent_request.headers
 
+    def _add_connection_response(
+        self, httpx_mock: HTTPXMock, connection_id: str
+    ) -> None:
+        httpx_mock.add_response(
+            method="GET",
+            status_code=200,
+            json={"id": connection_id, "name": "Test", "elementInstanceId": 1},
+        )
+
+    def test_invoke_activity_waits_longer_than_integration_service(
+        self,
+        httpx_mock: HTTPXMock,
+        service: ConnectionsService,
+        simple_activity_metadata: ActivityMetadata,
+    ) -> None:
+        """The activity call waits past Integration Service's ~120 s vendor budget, not the 30 s default."""
+        connection_id = "test-connection-123"
+        self._add_connection_response(httpx_mock, connection_id)
+        httpx_mock.add_response(method="POST", status_code=200, json={})
+
+        service.invoke_activity(
+            activity_metadata=simple_activity_metadata,
+            connection_id=connection_id,
+            activity_input={"body_field1": "x"},
+        )
+
+        connection_request, activity_request = httpx_mock.get_requests()
+        assert activity_request.extensions["timeout"]["read"] == 150.0
+        assert activity_request.extensions["timeout"]["connect"] == 30.0
+        assert connection_request.extensions["timeout"]["read"] == 30.0
+
+    def test_invoke_activity_read_timeout_not_retried(
+        self,
+        httpx_mock: HTTPXMock,
+        service: ConnectionsService,
+        simple_activity_metadata: ActivityMetadata,
+    ) -> None:
+        """A read timeout means Integration Service got the activity: it is not sent again."""
+        connection_id = "test-connection-123"
+        self._add_connection_response(httpx_mock, connection_id)
+        httpx_mock.add_exception(httpx.ReadTimeout("timed out"), method="POST")
+
+        with pytest.raises(httpx.ReadTimeout):
+            service.invoke_activity(
+                activity_metadata=simple_activity_metadata,
+                connection_id=connection_id,
+                activity_input={"body_field1": "x"},
+            )
+
+        assert [r.method for r in httpx_mock.get_requests()] == ["GET", "POST"]
+
+    @pytest.mark.anyio
+    async def test_invoke_activity_async_read_timeout_not_retried(
+        self,
+        httpx_mock: HTTPXMock,
+        service: ConnectionsService,
+        simple_activity_metadata: ActivityMetadata,
+    ) -> None:
+        connection_id = "test-connection-123"
+        self._add_connection_response(httpx_mock, connection_id)
+        httpx_mock.add_exception(httpx.ReadTimeout("timed out"), method="POST")
+
+        with pytest.raises(httpx.ReadTimeout):
+            await service.invoke_activity_async(
+                activity_metadata=simple_activity_metadata,
+                connection_id=connection_id,
+                activity_input={"body_field1": "x"},
+            )
+
+        activity_requests = [r for r in httpx_mock.get_requests() if r.method == "POST"]
+        assert len(activity_requests) == 1
+        assert activity_requests[0].extensions["timeout"]["read"] == 150.0
+
+    @pytest.mark.anyio
+    async def test_invoke_activity_async_connect_error_still_retried(
+        self,
+        httpx_mock: HTTPXMock,
+        service: ConnectionsService,
+        simple_activity_metadata: ActivityMetadata,
+    ) -> None:
+        """A failed connect never reached Integration Service, so it is still retried."""
+        connection_id = "test-connection-123"
+        self._add_connection_response(httpx_mock, connection_id)
+        httpx_mock.add_exception(httpx.ConnectTimeout("timed out"), method="POST")
+        httpx_mock.add_response(method="POST", status_code=200, json={"ok": True})
+
+        result = await service.invoke_activity_async(
+            activity_metadata=simple_activity_metadata,
+            connection_id=connection_id,
+            activity_input={"body_field1": "x"},
+        )
+
+        assert result == {"ok": True}
+        assert len([r for r in httpx_mock.get_requests() if r.method == "POST"]) == 2
+
+    def test_invoke_activity_retryable_status_still_retried(
+        self,
+        httpx_mock: HTTPXMock,
+        service: ConnectionsService,
+        simple_activity_metadata: ActivityMetadata,
+    ) -> None:
+        connection_id = "test-connection-123"
+        self._add_connection_response(httpx_mock, connection_id)
+        httpx_mock.add_response(
+            method="POST", status_code=429, headers={"retry-after": "0"}
+        )
+        httpx_mock.add_response(method="POST", status_code=200, json={"ok": True})
+
+        result = service.invoke_activity(
+            activity_metadata=simple_activity_metadata,
+            connection_id=connection_id,
+            activity_input={"body_field1": "x"},
+        )
+
+        assert result == {"ok": True}
+        assert len([r for r in httpx_mock.get_requests() if r.method == "POST"]) == 2
+
     def test_invoke_activity_with_body_fields(
         self,
         httpx_mock: HTTPXMock,