fix: prevent Actor log-streaming thread from crashing on stream timeout

src/apify_client/_streamed_log.py

@@ -5,23 +5,34 @@
 import re
 import threading
 from asyncio import Task
-from datetime import UTC, datetime, timedelta
+from datetime import UTC, datetime
 from threading import Thread
 from typing import TYPE_CHECKING, ClassVar, Self, cast
 
+import impit
+
 from apify_client._docs import docs_group
 
 if TYPE_CHECKING:
     from types import TracebackType
 
     from apify_client._resource_clients import LogClient, LogClientAsync
+    from apify_client.types import Timeout
 
 
 class StreamedLogBase:
     """Base class for streaming and buffering chunked Actor run logs."""
 
-    # Test related flag to enable propagation of logs to the `caplog` fixture during tests.
     _force_propagate = False
+    """Test related flag to enable propagation of logs to the `caplog` fixture during tests."""
+
+    _stream_timeout: ClassVar[Timeout] = 'no_timeout'
+    """Timeout for the log-stream long-poll request, which stays open for the whole Actor run.
+
+    impit applies its `timeout` to the whole request including the streamed body, so any bounded value truncates a
+    longer run mid-stream and raises `impit.TimeoutException` (#1040). `no_timeout` maps to impit's ~24h cap, which
+    is effectively unbounded for real runs and mirrors the JS client.
+    """
 
     def __init__(self, to_logger: logging.Logger, *, from_start: bool = True) -> None:
         if self._force_propagate:
@@ -90,10 +101,6 @@ class StreamedLog(StreamedLogBase):
     call `start` and `stop` manually. Obtain an instance via `RunClient.get_streamed_log`.
     """
 
-    # Caps how long `iter_bytes()` can block on a silent stream so `stop()` can unblock within
-    # this window instead of waiting for the long-polling default.
-    _read_timeout: ClassVar[timedelta] = timedelta(seconds=30)
-
     def __init__(self, log_client: LogClient, *, to_logger: logging.Logger, from_start: bool = True) -> None:
         """Initialize `StreamedLog`.
 
@@ -117,7 +124,8 @@ def start(self) -> Thread:
         if self._streaming_thread:
             raise RuntimeError('Streaming thread already active')
         self._stop_logging = False
-        self._streaming_thread = threading.Thread(target=self._stream_log)
+        # A daemon thread so a stream still blocked on a read can never hold up interpreter shutdown.
+        self._streaming_thread = threading.Thread(target=self._stream_log, daemon=True)
         self._streaming_thread.start()
         return self._streaming_thread
 
@@ -142,17 +150,25 @@ def __exit__(
         self.stop()
 
     def _stream_log(self) -> None:
-        with self._log_client.stream(raw=True, timeout=self._read_timeout) as log_stream:
-            if not log_stream:
-                return
-            try:
-                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)
+        try:
+            with self._log_client.stream(raw=True, timeout=self._stream_timeout) as log_stream:
+                if not log_stream:
+                    return
+                try:
+                    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)
+        except impit.TimeoutException:
+            # With `no_timeout` this fires only if the run outlives impit's ~24h cap or the connection stalls.
+            # The stream cannot continue, so warn and let the thread end instead of leaking a traceback (#1040).
+            self._to_logger.warning('Log streaming stopped: the log stream request timed out.')
+        except Exception:
+            # Any other failure in log redirection must not escape the background thread; log it instead.
+            self._to_logger.exception('Log redirection stopped due to unexpected error:')
 
 
 @docs_group('Other')
@@ -216,7 +232,7 @@ async def __aexit__(
 
     async def _stream_log(self) -> None:
         try:
-            async with self._log_client.stream(raw=True) as log_stream:
+            async with self._log_client.stream(raw=True, timeout=self._stream_timeout) as log_stream:
                 if not log_stream:
                     return
                 try:
@@ -225,6 +241,10 @@ async def _stream_log(self) -> None:
                 finally:
                     # Flush the last buffered part even if the task is cancelled by `stop()`.
                     self._log_buffer_content(include_last_part=True)
+        except impit.TimeoutException:
+            # As in `StreamedLog._stream_log`, impit's whole-request timeout on the long-lived stream is an
+            # expected terminal condition, not an error, so log a warning and end the task instead of a traceback.
+            self._to_logger.warning('Log streaming stopped: the log stream request timed out.')
         except Exception:
             # Exception in log redirection should not propagate further.
             self._to_logger.exception('Log redirection stopped due to unexpected error:')

tests/unit/test_logging.py

@@ -15,7 +15,7 @@
 from apify_client import ApifyClient, ApifyClientAsync
 from apify_client._logging import RedirectLogFormatter
 from apify_client._status_message_watcher import StatusMessageWatcherBase
-from apify_client._streamed_log import StreamedLog, StreamedLogBase
+from apify_client._streamed_log import StreamedLog, StreamedLogAsync, StreamedLogBase
 
 if TYPE_CHECKING:
     from collections.abc import Iterator
@@ -820,13 +820,13 @@ def generate_logs() -> Iterator[bytes]:
     assert any(_TAIL_SECOND_MESSAGE in m for m in messages), f'Buffered tail dropped on async stop(). Got: {messages}'
 
 
-def test_streamed_log_sync_stop_does_not_hang_on_silent_stream(
+def test_streamed_log_sync_stop_unblocks_on_finite_stream_timeout(
     httpserver: HTTPServer,
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    """Verify `stop()` returns promptly even when the underlying stream is silent (no chunks)."""
-    # Shorten the read timeout so the test doesn't wait for the production default.
-    monkeypatch.setattr(StreamedLog, '_read_timeout', timedelta(seconds=1))
+    """A finite `_stream_timeout` bounds how long `stop()` waits on a silent stream, since the blocking read cannot
+    otherwise be interrupted (the production default is `no_timeout`, so the test configures a short finite one)."""
+    monkeypatch.setattr(StreamedLog, '_stream_timeout', timedelta(seconds=1))
 
     release_server = threading.Event()
 
@@ -857,6 +857,143 @@ def generate_logs() -> Iterator[bytes]:
         stop_thread = threading.Thread(target=streamed_log.stop)
         stop_thread.start()
         stop_thread.join(timeout=5)
-        assert not stop_thread.is_alive(), 'stop() hangs when the underlying stream is silent'
+        assert not stop_thread.is_alive(), 'stop() did not unblock within the finite stream timeout'
     finally:
         release_server.set()
+
+
+@pytest.mark.usefixtures('propagate_stream_logs')
+def test_streamed_log_sync_does_not_leak_exception_on_stream_timeout(
+    caplog: LogCaptureFixture,
+    httpserver: HTTPServer,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The streaming thread ends quietly when the log-stream request hits its total timeout (regression #1040)."""
+    # impit enforces a whole-request timeout, so a still-running Actor whose run outlives the timeout makes
+    # `iter_bytes()` raise `impit.TimeoutException`. Shorten the timeout to trigger this quickly.
+    monkeypatch.setattr(StreamedLog, '_stream_timeout', timedelta(seconds=1))
+
+    release_server = threading.Event()
+
+    def _slow_handler(_request: Request) -> Response:
+        def generate_logs() -> Iterator[bytes]:
+            # Emit one complete line, then keep the connection open (as a running Actor would) past the
+            # client-side total timeout without sending anything more.
+            yield b'2025-05-13T07:24:12.588Z ACTOR: still running\n'
+            release_server.wait(timeout=30)
+
+        return Response(response=generate_logs(), status=200, mimetype='application/octet-stream')
+
+    httpserver.expect_request(
+        f'/v2/actor-runs/{_MOCKED_RUN_ID}/log', method='GET', query_string='stream=true&raw=true'
+    ).respond_with_handler(_slow_handler)
+    _register_run_and_actor_endpoints(httpserver)
+
+    api_url = httpserver.url_for('/').removesuffix('/')
+    run_client = ApifyClient(token='mocked_token', api_url=api_url).run(run_id=_MOCKED_RUN_ID)
+    streamed_log = run_client.get_streamed_log()
+    logger_name = f'apify.{_MOCKED_ACTOR_NAME}-{_MOCKED_RUN_ID}'
+
+    thread_exceptions: list[threading.ExceptHookArgs] = []
+    monkeypatch.setattr(threading, 'excepthook', thread_exceptions.append)
+
+    try:
+        with caplog.at_level(logging.DEBUG, logger=logger_name):
+            thread = streamed_log.start()
+            # Wait past the 1s total timeout so the streaming request fails inside the thread.
+            thread.join(timeout=5)
+            assert not thread.is_alive(), 'streaming thread did not end after the stream timed out'
+    finally:
+        release_server.set()
+        streamed_log.stop()
+
+    leaked = [args.exc_type.__name__ for args in thread_exceptions]
+    assert not leaked, f'streaming thread leaked an uncaught exception: {leaked}'
+    # The timeout is expected, so it must be swallowed quietly, not funnelled through the generic error handler.
+    error_records = [r for r in caplog.records if r.levelno >= logging.ERROR and 'Log redirection stopped' in r.message]
+    assert not error_records, f'sync thread logged an error on stream timeout: {[r.message for r in error_records]}'
+    # The line received before the timeout must still have been redirected.
+    assert any('ACTOR: still running' in record.message for record in caplog.records)
+
+
+@pytest.mark.usefixtures('propagate_stream_logs')
+def test_streamed_log_sync_requests_stream_with_no_timeout(
+    httpserver: HTTPServer,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The log stream is requested with `no_timeout`, so a long run is not truncated mid-stream (#1040)."""
+    httpserver.expect_request(
+        f'/v2/actor-runs/{_MOCKED_RUN_ID}/log', method='GET', query_string='stream=true&raw=true'
+    ).respond_with_data(b'2025-05-13T07:24:12.588Z ACTOR: done\n', content_type='application/octet-stream')
+    _register_run_and_actor_endpoints(httpserver)
+
+    api_url = httpserver.url_for('/').removesuffix('/')
+    run_client = ApifyClient(token='mocked_token', api_url=api_url).run(run_id=_MOCKED_RUN_ID)
+
+    # Capture the timeout the log stream is requested with. impit applies it to the whole request (body included),
+    # so anything but `no_timeout` would cut a long run off mid-stream, which is the root cause of #1040.
+    log_stream_timeouts: list[object] = []
+    original_call = run_client._http_client.call
+
+    def _recording_call(**kwargs: object) -> object:
+        if str(kwargs.get('url', '')).endswith('/log'):
+            log_stream_timeouts.append(kwargs.get('timeout'))
+        return original_call(**kwargs)
+
+    monkeypatch.setattr(run_client._http_client, 'call', _recording_call)
+
+    streamed_log = run_client.get_streamed_log()
+    thread = streamed_log.start()
+    thread.join(timeout=5)
+    streamed_log.stop()
+
+    assert log_stream_timeouts == ['no_timeout'], (
+        f'log stream requested with timeout={log_stream_timeouts}, expected no_timeout so long runs are not truncated'
+    )
+
+
+@pytest.mark.usefixtures('propagate_stream_logs')
+async def test_streamed_log_async_does_not_error_on_stream_timeout(
+    caplog: LogCaptureFixture,
+    httpserver: HTTPServer,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The async streaming task ends quietly on a stream-request timeout, matching the sync regression for #1040."""
+    monkeypatch.setattr(StreamedLogAsync, '_stream_timeout', timedelta(seconds=1))
+
+    release_server = threading.Event()
+
+    def _slow_handler(_request: Request) -> Response:
+        def generate_logs() -> Iterator[bytes]:
+            # Emit one complete line, then keep the connection open past the client-side total timeout.
+            yield b'2025-05-13T07:24:12.588Z ACTOR: still running\n'
+            release_server.wait(timeout=30)
+
+        return Response(response=generate_logs(), status=200, mimetype='application/octet-stream')
+
+    httpserver.expect_request(
+        f'/v2/actor-runs/{_MOCKED_RUN_ID}/log', method='GET', query_string='stream=true&raw=true'
+    ).respond_with_handler(_slow_handler)
+    _register_run_and_actor_endpoints(httpserver)
+
+    api_url = httpserver.url_for('/').removesuffix('/')
+    run_client = ApifyClientAsync(token='mocked_token', api_url=api_url).run(run_id=_MOCKED_RUN_ID)
+    streamed_log = await run_client.get_streamed_log()
+    logger_name = f'apify.{_MOCKED_ACTOR_NAME}-{_MOCKED_RUN_ID}'
+
+    try:
+        with caplog.at_level(logging.DEBUG, logger=logger_name):
+            task = streamed_log.start()
+            # The 1s total timeout fails the request inside the task; it must end on its own without our help.
+            done, _pending = await asyncio.wait({task}, timeout=5)
+            assert task in done, 'async streaming task did not end after the stream timed out'
+    finally:
+        release_server.set()
+        await streamed_log.stop()
+
+    assert not task.cancelled()
+    assert task.exception() is None, f'async streaming task raised on stream timeout: {task.exception()!r}'
+    error_records = [r for r in caplog.records if r.levelno >= logging.ERROR and 'Log redirection stopped' in r.message]
+    assert not error_records, f'async task logged an error on stream timeout: {[r.message for r in error_records]}'
+    # The line received before the timeout must still have been redirected.
+    assert any('ACTOR: still running' in record.message for record in caplog.records)