fix: Keep a failing background watcher from breaking a successful actor call

src/apify_client/_status_message_watcher.py

@@ -120,11 +120,19 @@ async def __aexit__(
         await self.stop()
 
     async def _log_changed_status_message(self) -> None:
-        while True:
-            run_data = await self._run_client.get()
-            if not self._log_run_data(run_data):
-                break
-            await asyncio.sleep(self._check_period)
+        try:
+            while True:
+                run_data = await self._run_client.get()
+                if not self._log_run_data(run_data):
+                    break
+                await asyncio.sleep(self._check_period)
+        except Exception as exc:
+            if self._run_client._http_client.is_timeout_error(exc):  # noqa: SLF001
+                # An expected timeout, so warn rather than leak a traceback.
+                self._to_logger.warning('Status message redirection stopped: the status request timed out.')
+            else:
+                # A failed poll must not escape into `stop` and surface as a failure of the run.
+                self._to_logger.exception('Status message redirection stopped due to unexpected error:')
 
 
 @docs_group('Other')
@@ -162,7 +170,8 @@ def start(self) -> Thread:
         if self._logging_thread:
             raise RuntimeError('Logging thread already active')
         self._stop_logging = False
-        self._logging_thread = threading.Thread(target=self._log_changed_status_message)
+        # A daemon thread, so a watcher still polling cannot hold up interpreter shutdown.
+        self._logging_thread = threading.Thread(target=self._log_changed_status_message, daemon=True)
         self._logging_thread.start()
         return self._logging_thread
 
@@ -189,9 +198,17 @@ def __exit__(
         self.stop()
 
     def _log_changed_status_message(self) -> None:
-        while True:
-            if not self._log_run_data(self._run_client.get()):
-                break
-            if self._stop_logging:
-                break
-            time.sleep(self._check_period)
+        try:
+            while True:
+                if not self._log_run_data(self._run_client.get()):
+                    break
+                if self._stop_logging:
+                    break
+                time.sleep(self._check_period)
+        except Exception as exc:
+            if self._run_client._http_client.is_timeout_error(exc):  # noqa: SLF001
+                # An expected timeout, so warn rather than leak a traceback.
+                self._to_logger.warning('Status message redirection stopped: the status request timed out.')
+            else:
+                # A failed poll must not escape the background thread.
+                self._to_logger.exception('Status message redirection stopped due to unexpected error:')

src/apify_client/_streamed_log.py

@@ -15,6 +15,7 @@
     from types import TracebackType
 
     from apify_client._resource_clients import LogClient, LogClientAsync
+    from apify_client.http_clients import HttpResponse
     from apify_client.types import Timeout
 
 
@@ -98,6 +99,13 @@ class StreamedLog(StreamedLogBase):
     call `start` and `stop` manually. Obtain an instance via `RunClient.get_streamed_log`.
     """
 
+    _stop_timeout_s: ClassVar[float] = 5
+    """Upper bound on how long `stop` waits for the streaming thread to finish.
+
+    Closing the response only ends the read on a transport that honours it - Impit does not - so without a bound
+    `stop` would wait for the next chunk, which on a quiet run may be hours away.
+    """
+
     def __init__(self, log_client: LogClient, *, to_logger: logging.Logger, from_start: bool = True) -> None:
         """Initialize `StreamedLog`.
 
@@ -111,14 +119,15 @@ def __init__(self, log_client: LogClient, *, to_logger: logging.Logger, from_sta
         super().__init__(to_logger=to_logger, from_start=from_start)
         self._log_client = log_client
         self._streaming_thread: Thread | None = None
+        self._log_stream: HttpResponse | None = None
         self._stop_logging = False
 
     def start(self) -> Thread:
         """Start the streaming thread.
 
         The caller is responsible for cleanup by calling the `stop` method when done.
         """
-        if self._streaming_thread:
+        if self._streaming_thread and self._streaming_thread.is_alive():
             raise RuntimeError('Streaming thread already active')
         self._stop_logging = False
         # A daemon thread so a stream still blocked on a read can never hold up interpreter shutdown.
@@ -127,13 +136,28 @@ def start(self) -> Thread:
         return self._streaming_thread
 
     def stop(self) -> None:
-        """Signal the streaming thread to stop logging and wait for it to finish."""
+        """Signal the streaming thread to stop logging and wait up to `_stop_timeout_s` for it to finish.
+
+        A thread that outlives the wait is a daemon with `_stop_logging` set, so it exits after at most one more chunk.
+        Its handle is kept while it is alive, so `start` cannot revive it beside a second thread on the same buffer.
+        """
         if not self._streaming_thread:
             raise RuntimeError('Streaming thread is not active')
         self._stop_logging = True
-        self._streaming_thread.join()
-        self._streaming_thread = None
-        self._stop_logging = False
+        # Read once; the streaming thread clears the attribute as soon as the stream ends.
+        log_stream = self._log_stream
+        if log_stream is not None:
+            try:
+                log_stream.close()
+            except Exception:
+                # A failing `close` in a custom transport must not fail the caller.
+                self._to_logger.exception('Closing the log stream failed:')
+        self._streaming_thread.join(timeout=self._stop_timeout_s)
+        if self._streaming_thread.is_alive():
+            # Otherwise log messages arriving after `stop` returned have no explanation.
+            self._to_logger.debug('Log streaming thread outlived the stop timeout; it ends after the next chunk.')
+        else:
+            self._streaming_thread = None
 
     def __enter__(self) -> Self:
         """Start the streaming thread within the context. Exiting the context will finish the streaming thread."""
@@ -151,15 +175,25 @@ def _stream_log(self) -> None:
             with self._log_client.stream(raw=True, timeout=self._stream_timeout) as log_stream:
                 if not log_stream:
                     return
+                # Published so `stop` can close the response.
+                self._log_stream = log_stream
                 try:
+                    # `stop` may have run before the response existed for it to close.
+                    if self._stop_logging:
+                        return
                     for data in log_stream.iter_bytes():
                         self._process_new_data(data)
                         if self._stop_logging:
                             break
                 finally:
+                    self._log_stream = None
                     # Flush the last buffered part even if the read timed out or was stopped.
                     self._log_buffer_content(include_last_part=True)
         except Exception as exc:
+            if self._stop_logging:
+                # Expected during a stop, but this also catches a failed flush of the buffered tail, so report it.
+                self._to_logger.debug('Log streaming stopped while `stop` was in progress: %r', exc)
+                return
             if self._log_client._http_client.is_timeout_error(exc):  # noqa: SLF001
                 # The stream cannot continue, so warn and let the thread end instead of leaking a traceback.
                 self._to_logger.warning('Log streaming stopped: the log stream request timed out.')

tests/integration/test_log.py

@@ -2,15 +2,23 @@
 
 from __future__ import annotations
 
+import asyncio
+import logging
+import threading
 from contextlib import AbstractAsyncContextManager, AbstractContextManager
 from typing import TYPE_CHECKING
 
 from .._utils import maybe_await
 from apify_client._models import ListOfBuilds, Run
+from apify_client._resource_clients import RunClient, RunClientAsync
 from apify_client.http_clients import HttpResponse
 
 if TYPE_CHECKING:
+    import pytest
+    from _pytest.logging import LogCaptureFixture
+
     from apify_client import ApifyClient, ApifyClientAsync
+    from apify_client.types import Timeout
 
 # Use a simple, fast public actor for testing
 HELLO_WORLD_ACTOR = 'apify/hello-world'
@@ -99,3 +107,49 @@ async def test_log_stream_from_run(client: ApifyClient | ApifyClientAsync, *, is
                 assert len(content) > 0
     finally:
         await maybe_await(run_client.delete())
+
+
+async def test_actor_call_returns_run_when_status_poll_fails(
+    client: ApifyClient | ApifyClientAsync,
+    caplog: LogCaptureFixture,
+    monkeypatch: pytest.MonkeyPatch,
+    *,
+    is_async: bool,
+) -> None:
+    """A failing status poll is reported to the logger and leaves `call` returning the run the platform finished."""
+    logger = logging.getLogger(f'test-status-poll-failure-{"async" if is_async else "sync"}')
+
+    # `call` polls the run status from a background task or thread and awaits the run itself on the main one, so
+    # failing only the background polls leaves the rest of `call` talking to the real API.
+    if is_async:
+        main_task = asyncio.current_task()
+        original_get_async = RunClientAsync.get
+
+        async def failing_get_async(self: RunClientAsync, *, timeout: Timeout = 'short') -> Run | None:
+            if asyncio.current_task() is main_task:
+                return await original_get_async(self, timeout=timeout)
+            raise RuntimeError('Simulated status poll failure')
+
+        monkeypatch.setattr(RunClientAsync, 'get', failing_get_async)
+    else:
+        original_get = RunClient.get
+
+        def failing_get(self: RunClient, *, timeout: Timeout = 'short') -> Run | None:
+            if threading.current_thread() is threading.main_thread():
+                return original_get(self, timeout=timeout)
+            raise RuntimeError('Simulated status poll failure')
+
+        monkeypatch.setattr(RunClient, 'get', failing_get)
+
+    with caplog.at_level(logging.DEBUG, logger=logger.name):
+        run = await maybe_await(client.actor(HELLO_WORLD_ACTOR).call(logger=logger))
+
+    assert isinstance(run, Run)
+    assert run.status == 'SUCCEEDED'
+    assert any(
+        record.levelno == logging.ERROR
+        and record.message == 'Status message redirection stopped due to unexpected error:'
+        for record in caplog.records
+    )
+
+    await maybe_await(client.run(run.id).delete())