fix: dedicated http client for governance telemetry dispatcher

Author: radu-mocanu

packages/uipath-platform/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "uipath-platform"
-version = "0.2.12"
+version = "0.2.13"
 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/governance/_live_track_event_dispatcher.py

@@ -22,7 +22,11 @@
   assumes it owns the provider's async HTTP path — nothing else in
   the process should await ``track_event_async`` (or any other
   ``*_async`` method on the same underlying service) on a *different*
-  loop. See "one dispatcher per provider" below.
+  loop. In particular, the provider passed in must be backed by a
+  service whose async client has not already served a request on
+  another loop (e.g. a governance policy fetch on the CLI's main
+  loop) — hand the dispatcher a *dedicated* provider. See "one
+  dispatcher per provider" below.
 
 - **Backpressure.** A ``BoundedSemaphore`` caps in-flight coroutines;
   submissions that exceed the cap are dropped with a warning so
@@ -74,6 +78,19 @@ class LiveTrackEventDispatcher:
 
     _DEFAULT_MAX_INFLIGHT = 40
 
+    # Total wall-clock ceiling for a single dispatched call. The platform
+    # call retries (up to 5 attempts, honoring ``Retry-After`` up to 120s)
+    # and httpx's 30s timeout is per-phase, not total — so one degraded
+    # call could otherwise run for minutes, pinning an in-flight slot and
+    # stalling ``shutdown`` drain. This caps every dispatched call.
+    _PER_CALL_DEADLINE_SECONDS = 60.0
+
+    # Max wait for the background loop thread to signal readiness. The
+    # thread only fails to signal if it never starts (e.g. OS
+    # thread-creation failure); the caller treats a construction failure as
+    # "governance unavailable" (fail open) rather than blocking forever.
+    _LOOP_START_TIMEOUT_SECONDS = 5.0
+
     def __init__(
         self,
         provider: UiPathPlatformGovernanceProvider,
@@ -103,6 +120,10 @@ def __init__(
         self._shutdown_event = threading.Event()
         self._futures_lock = threading.Lock()
         self._futures: set[concurrent.futures.Future[None]] = set()
+        # Guards warn-once for swallowed dispatch failures. Only ever read
+        # or written on the background loop thread (inside ``_run``), so no
+        # lock is needed.
+        self._dispatch_failure_logged = False
 
         self._loop = asyncio.new_event_loop()
         self._loop_ready = threading.Event()
@@ -113,13 +134,31 @@ def __init__(
         )
         self._loop_thread.start()
         # Block until the loop is running so the first ``dispatch`` cannot
-        # race with startup and hit "loop not running" errors.
-        self._loop_ready.wait()
+        # race with startup and hit "loop not running" errors. Bounded so a
+        # thread that never starts surfaces as a construction failure the
+        # caller can fall back on, instead of hanging the process forever.
+        if not self._loop_ready.wait(timeout=self._LOOP_START_TIMEOUT_SECONDS):
+            # Signal the (possibly late-starting) loop thread to abort and
+            # close its own loop. NEVER close it here, across threads — that
+            # would race the thread's ``run_forever`` (closing a running
+            # loop, or running an already-closed one). See ``_run_loop``.
+            self._shutdown_event.set()
+            raise RuntimeError(
+                "governance track-event loop failed to start within "
+                f"{self._LOOP_START_TIMEOUT_SECONDS}s"
+            )
 
     def _run_loop(self) -> None:
         """Body of the background loop thread — runs until ``shutdown``."""
         asyncio.set_event_loop(self._loop)
         self._loop_ready.set()
+        # If construction already gave up waiting for readiness (the
+        # loop-start-timeout path set ``_shutdown_event`` before this thread
+        # got scheduled), abort now and close the loop HERE — on the thread
+        # that owns it — so ``__init__`` never closes it across threads.
+        if self._shutdown_event.is_set():
+            self._loop.close()
+            return
         try:
             self._loop.run_forever()
         finally:
@@ -170,11 +209,13 @@ def dispatch(
           raises ``RuntimeError`` if the loop is stopped/closed
           (late-firing atexit path); the dispatcher rolls back the
           semaphore slot, closes the coroutine, and logs at debug.
-        - **Coroutine exception**: the provider's HTTP call may raise
-          for any reason (serialization, 5xx, transport). ``_run``
-          catches, logs at debug with ``exc_info=True``, and the
-          done-callback observes the future to suppress asyncio's
-          "exception was never retrieved" warning.
+        - **Coroutine exception / deadline**: the provider's HTTP call may
+          raise for any reason (serialization, 5xx, transport) or exceed
+          ``_PER_CALL_DEADLINE_SECONDS``. ``_run`` catches both, logs the
+          first such failure at warning (so silent telemetry loss is
+          visible) and the rest at debug, and the done-callback observes
+          the future to suppress asyncio's "exception was never retrieved"
+          warning.
         """
         if self._shutdown_event.is_set():
             logger.debug(
@@ -218,15 +259,47 @@ async def _run(
         data: dict[str, Any] | None,
         operation_id: str | None,
     ) -> None:
-        """Coroutine body — the async HTTP call itself."""
+        """Coroutine body — the async HTTP call itself, under a total deadline."""
         try:
-            await self._provider.track_event_async(
-                event_name=event_name,
-                data=data,
-                operation_id=operation_id,
+            async with asyncio.timeout(self._PER_CALL_DEADLINE_SECONDS):
+                await self._provider.track_event_async(
+                    event_name=event_name,
+                    data=data,
+                    operation_id=operation_id,
+                )
+        except TimeoutError:
+            # The platform call outran the per-call deadline (retries +
+            # Retry-After can otherwise run for minutes). Drop it rather
+            # than let a degraded backend pin an in-flight slot.
+            self._log_dispatch_failure(
+                "track_event exceeded the %.0fs deadline; dropped (event_name=%s)",
+                self._PER_CALL_DEADLINE_SECONDS,
+                event_name,
             )
         except Exception as exc:  # noqa: BLE001 - fire-and-forget contract
-            logger.debug("Failed to dispatch track_event: %s", exc, exc_info=True)
+            self._log_dispatch_failure(
+                "Failed to dispatch track_event (event_name=%s): %s",
+                event_name,
+                exc,
+                exc_info=True,
+            )
+
+    def _log_dispatch_failure(
+        self, msg: str, *args: object, exc_info: bool = False
+    ) -> None:
+        """Log a swallowed dispatch failure — first at warning, then debug.
+
+        Dispatch failures are silent to the caller by the fire-and-forget
+        contract, so the first one is surfaced at warning to make telemetry
+        loss visible; subsequent failures drop to debug so a sustained-down
+        backend cannot flood the logs. Only ever called on the background
+        loop thread, so the flag read/write needs no lock.
+        """
+        if not self._dispatch_failure_logged:
+            self._dispatch_failure_logged = True
+            logger.warning(msg, *args, exc_info=exc_info)
+        else:
+            logger.debug(msg, *args, exc_info=exc_info)
 
     def _on_future_done(self, future: concurrent.futures.Future[None]) -> None:
         """Observe the future, drop it from the pending set, release the slot.
@@ -253,7 +326,7 @@ def _on_future_done(self, future: concurrent.futures.Future[None]) -> None:
                 self._futures.discard(future)
             self._inflight.release()
 
-    def shutdown(self, *, wait: bool = True, timeout: float = 30.0) -> None:
+    def shutdown(self, *, wait: bool = True, timeout: float = 5.0) -> None:
         """Stop accepting new submissions; optionally drain pending, then stop the loop.
 
         Call at process exit to avoid losing in-flight telemetry.
@@ -267,7 +340,10 @@ def shutdown(self, *, wait: bool = True, timeout: float = 30.0) -> None:
                 teardown path.
             timeout: Maximum seconds to wait for pending coroutines
                 when ``wait=True``. Coroutines still in flight after
-                the timeout are cancelled by loop teardown.
+                the timeout are cancelled by loop teardown. The default
+                is deliberately short: the only caller is a CLI exit path,
+                where a long drain against a degraded backend would just
+                stall process teardown (telemetry is best-effort).
         """
         if self._shutdown_event.is_set():
             return

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

@@ -16,6 +16,7 @@
 from __future__ import annotations
 
 import asyncio
+import inspect
 import logging
 import threading
 import time
@@ -464,37 +465,42 @@ def test_shutdown_swallows_when_loop_already_stopped(provider: MagicMock) -> Non
     dispatcher.shutdown()
 
 
-def test_worker_exception_is_logged_at_debug(
+def test_worker_exception_warns_once_then_debug(
     provider: MagicMock,
     dispatcher: LiveTrackEventDispatcher,
     caplog: pytest.LogCaptureFixture,
 ) -> None:
-    """Covers the ``except Exception`` branch inside ``_run``.
-
-    Complements :func:`test_worker_exception_does_not_propagate` by
-    asserting the log record (with ``exc_info``) actually fires, so
-    coverage records the debug-log line inside the coroutine.
+    """Covers the ``except Exception`` branch inside ``_run`` and the
+    warn-once policy.
+
+    Swallowed coroutine exceptions are silent to the caller by the
+    fire-and-forget contract, so the FIRST one is surfaced at warning (to
+    make otherwise-invisible telemetry loss visible) and the rest drop to
+    debug (so a sustained-down backend can't flood the logs). Both carry
+    ``exc_info`` so operators can trace the failure.
     """
     provider.track_event_async.side_effect = ValueError("bad payload")
 
     with caplog.at_level(logging.DEBUG, logger=_DISPATCHER_LOGGER):
-        dispatcher.dispatch(event_name="agent.bad")
-        # Wait for the coroutine to run AND the callback to fire so
-        # coverage collects the except-branch lines from the loop thread.
+        dispatcher.dispatch(event_name="agent.bad.1")
         assert _wait_for(lambda: provider.track_event_async.await_count >= 1)
-        # Small sleep to let the callback finalize (release semaphore,
-        # drop from set) — otherwise coverage may race the callback.
+        time.sleep(0.05)
+        dispatcher.dispatch(event_name="agent.bad.2")
+        assert _wait_for(lambda: provider.track_event_async.await_count >= 2)
         time.sleep(0.05)
 
-    matching = [
-        r
-        for r in caplog.records
-        if "Failed to dispatch track_event" in r.message and r.levelno == logging.DEBUG
+    failures = [
+        r for r in caplog.records if "Failed to dispatch track_event" in r.message
     ]
-    assert matching, "expected a debug log for the swallowed exception"
-    # exc_info is attached so operators can trace the failure.
-    assert matching[0].exc_info is not None
-    assert isinstance(matching[0].exc_info[1], ValueError)
+    warnings = [r for r in failures if r.levelno == logging.WARNING]
+    debugs = [r for r in failures if r.levelno == logging.DEBUG]
+
+    # Exactly one warning (warn-once), and at least one subsequent debug.
+    assert len(warnings) == 1, f"expected one warning, got {len(warnings)}"
+    assert debugs, "expected subsequent failures logged at debug"
+    # exc_info attached on the surfaced warning.
+    assert warnings[0].exc_info is not None
+    assert isinstance(warnings[0].exc_info[1], ValueError)
 
 
 # ---------------------------------------------------------------------------
@@ -579,3 +585,121 @@ async def _never_finish(**_: object) -> None:
     # few seconds — a hang here would freeze the whole test suite.
     assert elapsed < 3.0, f"shutdown took {elapsed:.3f}s with timeout=0.1s"
     assert not dispatcher._loop_thread.is_alive()
+
+
+def test_shutdown_default_timeout_is_short_for_cli_exit() -> None:
+    """The drain default must be short.
+
+    The only caller is a CLI exit path; a long drain against a degraded
+    backend would just stall process teardown (telemetry is best-effort).
+    Guards against silent drift back to a large default.
+    """
+    default = (
+        inspect.signature(LiveTrackEventDispatcher.shutdown)
+        .parameters["timeout"]
+        .default
+    )
+    assert default == 5.0
+
+
+# ---------------------------------------------------------------------------
+# per-call deadline
+# ---------------------------------------------------------------------------
+
+
+def test_per_call_deadline_drops_slow_call(
+    provider: MagicMock,
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A single dispatched call that outruns the per-call deadline is
+    cancelled and dropped — it must not pin its in-flight slot, and the
+    caller must never see an error.
+
+    Without the deadline, one degraded call (5 retries × 30s + Retry-After
+    backoff) could hold a slot for minutes and stall shutdown drain.
+    """
+    monkeypatch.setattr(LiveTrackEventDispatcher, "_PER_CALL_DEADLINE_SECONDS", 0.1)
+
+    cancelled = threading.Event()
+
+    async def _too_slow(**_: object) -> None:
+        try:
+            await asyncio.sleep(5.0)  # far beyond the 0.1s deadline
+        except asyncio.CancelledError:
+            cancelled.set()
+            raise
+
+    provider.track_event_async.side_effect = _too_slow
+
+    dispatcher = LiveTrackEventDispatcher(provider, max_inflight=1)
+    try:
+        with caplog.at_level(logging.WARNING, logger=_DISPATCHER_LOGGER):
+            dispatcher.dispatch(event_name="slow.event")
+            # The 0.1s deadline fires long before the 5s sleep would.
+            assert cancelled.wait(timeout=2.0), (
+                "coroutine was not cancelled at the deadline"
+            )
+            # The in-flight slot is released once the dropped call finalizes,
+            # so it can be re-acquired.
+            reacquired = _wait_for(lambda: dispatcher._inflight.acquire(blocking=False))
+            assert reacquired, "in-flight slot not released after deadline drop"
+            dispatcher._inflight.release()
+
+        deadline_logs = [
+            r
+            for r in caplog.records
+            if "exceeded the" in r.message and "deadline" in r.message
+        ]
+        assert deadline_logs, "expected a deadline-exceeded log"
+    finally:
+        dispatcher.shutdown()
+
+
+# ---------------------------------------------------------------------------
+# construction: bounded wait for the loop thread
+# ---------------------------------------------------------------------------
+
+
+def test_late_loop_start_raises_and_thread_closes_its_own_loop(
+    provider: MagicMock,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A loop thread that starts AFTER the readiness deadline must:
+
+    1. cause construction to fail fast (fail open — the bootstrap treats a
+       dispatcher construction failure as "governance unavailable", so it
+       can never hang the CLI), and
+    2. abort cleanly by closing its OWN loop when it finally runs — the
+       loop is never closed across threads (which would race
+       ``run_forever``: closing a running loop, or running a closed one).
+
+    Regression guard for the loop-affinity-review finding: ``__init__`` must
+    only *signal* the abort, and ``_run_loop`` must do the close.
+    """
+    monkeypatch.setattr(LiveTrackEventDispatcher, "_LOOP_START_TIMEOUT_SECONDS", 0.1)
+
+    gate = threading.Event()
+    real_run_loop = LiveTrackEventDispatcher._run_loop
+    captured: dict[str, LiveTrackEventDispatcher] = {}
+
+    def _delayed_run_loop(self: LiveTrackEventDispatcher) -> None:
+        # Hold the thread past the readiness deadline so __init__ times out
+        # first, THEN run the real loop body (which must hit the abort path).
+        captured["self"] = self
+        gate.wait(timeout=2.0)
+        real_run_loop(self)
+
+    monkeypatch.setattr(LiveTrackEventDispatcher, "_run_loop", _delayed_run_loop)
+
+    with pytest.raises(RuntimeError, match="failed to start"):
+        LiveTrackEventDispatcher(provider)
+
+    # __init__ has raised and signalled abort. Release the late thread into
+    # the REAL _run_loop: it must observe the abort flag, close its own loop,
+    # and exit — no run_forever, no leaked running loop, no cross-thread close.
+    dispatcher = captured["self"]
+    gate.set()
+    dispatcher._loop_thread.join(timeout=2.0)
+    assert not dispatcher._loop_thread.is_alive(), "late loop thread did not exit"
+    assert dispatcher._loop.is_closed(), "aborting thread did not close its own loop"

packages/uipath-platform/uv.lock

@@ -1,6 +1,6 @@
 [project]
 name = "uipath"
-version = "2.13.13"
+version = "2.13.14"
 description = "Python SDK and CLI for UiPath Platform, enabling programmatic interaction with automation services, process management, and deployment tools."
 readme = { file = "README.md", content-type = "text/markdown" }
 requires-python = ">=3.11"