fix: Keep a failing background watcher from breaking a successful actor call (#1027)

### Description

`actor.call()` with the default `logger='default'` could raise on a run
that had already succeeded. A status poll that failed after retries left
the exception on the watcher task; `stop()` then cancelled an
already-finished task and `await`ed it, re-raising past `except
asyncio.CancelledError`. The sync twin leaked the same failure out of
its thread.

- Both watcher loops now use the handler `StreamedLog._stream_log`
already had: warn on `is_timeout_error`, `logger.exception` otherwise. A
failed poll gets reported and the run's own result stands.
- The sync watcher's thread is a daemon, so a watcher whose loop never
terminates cannot hold up interpreter shutdown.
- `StreamedLog.stop()` closes the stream response and caps its join at
`_stop_timeout_s` (5s), where it used to join unbounded on a stream that
may not speak for hours.
- `start()` on both sync classes checks `is_alive()`, so a handle left
by a thread that outlived its stop does not block a restart.

The close alone is not enough. On Impit 0.13.2 `Response.close()`
returns immediately and sets `is_closed=True`, yet a reader blocked in
`iter_bytes()` stays blocked, and closing the `Client` does not release
it either. The close is kept because it is correct for a custom HTTP
client, and the capped join is what makes `stop()` return.

### Why `stop()` is bounded this way

- #825 bounded `stop()` by requesting the log stream with a 30s read
timeout.
- #944 removed that: Impit applies `timeout` to the whole request,
streamed body included, so a run logging past the bound was cut off
mid-stream with `impit.TimeoutException` (#945). The stream has used
`no_timeout` since, with the unbounded `stop()` recorded as a known
limitation.
- This PR caps the join and closes the response, leaving the stream
request alone. `_stream_timeout` stays `no_timeout`, still asserted by
`test_streamed_log_sync_requests_stream_with_no_timeout`, so #945 cannot
regress. What it fixes is the case #944 left open: a manual `stop()` on
a stream gone quiet.

Trade-off: a thread outliving the 5s cap keeps running as a daemon, so
`stop()` can return before the tail is flushed. On the `actor.call()`
path the run's EOF ends the thread first, measured at 0.000s over
repeated `apify/hello-world` runs.

### Test plan

- 8 unit tests and an integration test that fails a real run's
background status polls, each red before the fix.

*✍️ Drafted by Claude Code*

Author: vdusek

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')
@@ -159,10 +167,11 @@ def start(self) -> Thread:
 
         The caller is responsible for cleanup by calling the `stop` method when done.
         """
-        if self._logging_thread:
+        if self._logging_thread and self._logging_thread.is_alive():
             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, which neither Impit nor HTTPX does, 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,29 @@ 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,
+        and only then does its buffered tail reach the logger. 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 +176,30 @@ 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:
-                    # Flush the last buffered part even if the read timed out or was stopped.
-                    self._log_buffer_content(include_last_part=True)
+                    self._log_stream = None
+                    try:
+                        # Flush the last buffered part even if the read timed out or was stopped.
+                        self._log_buffer_content(include_last_part=True)
+                    except Exception:
+                        # A truncated stream leaves an undecodable tail, which is worth a traceback even while a stop
+                        # is in progress.
+                        self._to_logger.exception('Log redirection stopped due to unexpected error:')
         except Exception as exc:
+            if self._stop_logging:
+                # `stop` closed the stream out from under the read, so the failure is expected.
+                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.')
@@ -236,8 +276,13 @@ async def _stream_log(self) -> None:
                     async for data in log_stream.aiter_bytes():
                         self._process_new_data(data)
                 finally:
-                    # Flush the last buffered part even if the task is cancelled by `stop()`.
-                    self._log_buffer_content(include_last_part=True)
+                    try:
+                        # Flush the last buffered part even if the task is cancelled by `stop()`.
+                        self._log_buffer_content(include_last_part=True)
+                    except Exception:
+                        # A truncated stream leaves an undecodable tail. Keeping the failure here also keeps the
+                        # cancellation `stop` raised propagating, so the task ends up cancelled as asyncio expects.
+                        self._to_logger.exception('Log redirection stopped due to unexpected error:')
         except Exception as exc:
             if self._log_client._http_client.is_timeout_error(exc):  # noqa: SLF001
                 # A timeout on the long-lived stream is an expected terminal condition, not an error.

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,50 @@ 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)
+    try:
+        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
+        )
+    finally:
+        await maybe_await(client.run(run.id).delete())