diff --git a/src/apify_client/_streamed_log.py b/src/apify_client/_streamed_log.py index 3e693e2c..bdc89d8a 100644 --- a/src/apify_client/_streamed_log.py +++ b/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:') diff --git a/tests/unit/test_logging.py b/tests/unit/test_logging.py index 9fe56fa3..a5148696 100644 --- a/tests/unit/test_logging.py +++ b/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)