diff --git a/src/apify_client/_status_message_watcher.py b/src/apify_client/_status_message_watcher.py index a1bf1e38..030db953 100644 --- a/src/apify_client/_status_message_watcher.py +++ b/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:') diff --git a/src/apify_client/_streamed_log.py b/src/apify_client/_streamed_log.py index 428caa12..a9bb8c5a 100644 --- a/src/apify_client/_streamed_log.py +++ b/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,6 +119,7 @@ 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: @@ -118,7 +127,7 @@ def start(self) -> 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.') diff --git a/tests/integration/test_log.py b/tests/integration/test_log.py index df687e2f..b497ccc7 100644 --- a/tests/integration/test_log.py +++ b/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()) diff --git a/tests/unit/test_logging.py b/tests/unit/test_logging.py index 35b10e2e..03926b16 100644 --- a/tests/unit/test_logging.py +++ b/tests/unit/test_logging.py @@ -7,7 +7,7 @@ import time from datetime import datetime, timedelta from typing import TYPE_CHECKING -from unittest.mock import patch +from unittest.mock import Mock, patch import pytest from werkzeug import Request, Response @@ -835,14 +835,15 @@ def test_streamed_log_sync_stop_unblocks_on_finite_stream_timeout( """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)) + # Well above the stream timeout, so `stop`'s own join bound cannot end the wait first and mask a regression. + monkeypatch.setattr(StreamedLog, '_stop_timeout_s', 30) release_server = threading.Event() def _silent_handler(_request: Request) -> Response: def generate_logs() -> Iterator[bytes]: - # Yield an empty chunk so werkzeug flushes headers and the client sees a streaming - # response; then block without emitting any log data. - yield b'' + # One complete line, then silence, like a running Actor that stops logging. + yield b'2025-05-13T07:24:12.588Z ACTOR: going quiet\n' release_server.wait(timeout=30) return Response(response=generate_logs(), status=200, mimetype='application/octet-stream') @@ -858,8 +859,11 @@ def generate_logs() -> Iterator[bytes]: streamed_log.start() try: - # Give the streaming thread time to start and block inside iter_bytes. - time.sleep(0.3) + # The buffered line proves the thread is inside the read loop; a fixed sleep could end before it gets there. + deadline = time.monotonic() + 5 + while not streamed_log._stream_buffer and time.monotonic() < deadline: + time.sleep(0.01) + assert streamed_log._stream_buffer, 'streaming thread never reached the blocking read' # Call stop() from a helper thread so the test cannot hang indefinitely if the fix regresses. stop_thread = threading.Thread(target=streamed_log.stop) @@ -1015,6 +1019,188 @@ def generate_logs() -> Iterator[bytes]: assert any('ACTOR: still running' in record.message for record in caplog.records) +_POLL_FAILURE_FINAL_SLEEP_S = 4 +"""Long enough for the once-per-second watcher poll to reach the rejecting endpoint.""" + + +@pytest.fixture +def mock_api_failing_status_poll(httpserver: HTTPServer) -> None: + """Set up the endpoints `call` needs, with the status poll rejected once the watcher is its only caller.""" + status_generator = StatusResponseGenerator() + running_run = status_generator._create_minimal_run_data('Initial message', 'RUNNING', is_terminal=False) + finished_run = status_generator._create_minimal_run_data('Final message', 'SUCCEEDED', is_terminal=True) + # `call` requests the log stream only after both of its setup status requests return, so every plain status + # request after that is a watcher poll. A count cannot tell them apart: the watcher polls before the second. + setup_done = threading.Event() + + def _status_handler(request: Request) -> Response: + if 'waitForFinish' in request.args: + # `wait_for_finish` keeps succeeding, so the run reads as a success. + return Response(response=json.dumps({'data': finished_run}), status=200, mimetype='application/json') + if setup_done.is_set(): + return Response( + response=json.dumps({'error': {'type': 'insufficient-permissions', 'message': 'Poll rejected'}}), + status=403, + mimetype='application/json', + ) + return Response(response=json.dumps({'data': running_run}), status=200, mimetype='application/json') + + def _log_handler(request: Request) -> Response: + setup_done.set() + return _streaming_log_handler(request) + + # Registered before `_register_run_and_actor_endpoints`, which also covers the run endpoint - first match wins. + httpserver.expect_request(f'/v2/actor-runs/{_MOCKED_RUN_ID}', method='GET').respond_with_handler(_status_handler) + _register_run_and_actor_endpoints(httpserver) + httpserver.expect_request(f'/v2/actors/{_MOCKED_ACTOR_ID}/runs', method='POST').respond_with_json( + {'data': running_run} + ) + httpserver.expect_request( + f'/v2/actor-runs/{_MOCKED_RUN_ID}/log', method='GET', query_string='stream=true&raw=true' + ).respond_with_handler(_log_handler) + + +@pytest.mark.usefixtures('mock_api_failing_status_poll', 'propagate_stream_logs') +async def test_actor_call_returns_run_when_status_poll_fails_async( + caplog: LogCaptureFixture, + httpserver: HTTPServer, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A failing status poll is logged by the watcher instead of surfacing as a failure of the finished run.""" + monkeypatch.setattr(StatusMessageWatcherBase, '_final_sleep_time_s', _POLL_FAILURE_FINAL_SLEEP_S) + + api_url = httpserver.url_for('/').removesuffix('/') + actor_client = ApifyClientAsync(token='mocked_token', api_url=api_url).actor(actor_id=_MOCKED_ACTOR_ID) + logger_name = f'apify.{_MOCKED_ACTOR_NAME} runId:{_MOCKED_RUN_ID}' + + with caplog.at_level(logging.DEBUG, logger=logger_name): + run = await actor_client.call() + + assert run is not None + assert run.status == 'SUCCEEDED' + assert any('Status message redirection stopped' in record.message for record in caplog.records) + + +@pytest.mark.usefixtures('mock_api_failing_status_poll', 'propagate_stream_logs') +def test_actor_call_returns_run_when_status_poll_fails_sync( + caplog: LogCaptureFixture, + httpserver: HTTPServer, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A failing status poll is logged by the watcher thread instead of leaking out of it.""" + monkeypatch.setattr(StatusMessageWatcherBase, '_final_sleep_time_s', _POLL_FAILURE_FINAL_SLEEP_S) + + thread_exceptions: list[threading.ExceptHookArgs] = [] + monkeypatch.setattr(threading, 'excepthook', thread_exceptions.append) + + api_url = httpserver.url_for('/').removesuffix('/') + actor_client = ApifyClient(token='mocked_token', api_url=api_url).actor(actor_id=_MOCKED_ACTOR_ID) + logger_name = f'apify.{_MOCKED_ACTOR_NAME} runId:{_MOCKED_RUN_ID}' + + with caplog.at_level(logging.DEBUG, logger=logger_name): + run = actor_client.call() + + assert run is not None + assert run.status == 'SUCCEEDED' + leaked = [args.exc_type.__name__ for args in thread_exceptions] + assert not leaked, f'polling thread leaked an uncaught exception: {leaked}' + assert any('Status message redirection stopped' in record.message for record in caplog.records) + + +@pytest.mark.usefixtures('mock_api') +def test_sync_watcher_thread_is_daemon(httpserver: HTTPServer) -> None: + """The polling thread is a daemon, so a watcher still polling cannot hold up interpreter shutdown.""" + api_url = httpserver.url_for('/').removesuffix('/') + run_client = ApifyClient(token='mocked_token', api_url=api_url).run(run_id=_MOCKED_RUN_ID) + watcher = run_client.get_status_message_watcher(check_period=timedelta(seconds=0)) + + thread = watcher.start() + try: + assert thread.daemon + finally: + watcher.stop() + + +def test_streamed_log_sync_stop_returns_on_silent_stream( + httpserver: HTTPServer, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """`stop` returns within its bound on a silent stream, and the log can be started again once the thread ends.""" + monkeypatch.setattr(StreamedLog, '_stop_timeout_s', 1) + + release_server = threading.Event() + + def _silent_handler(_request: Request) -> Response: + def generate_logs() -> Iterator[bytes]: + # One complete line, then silence, like a running Actor that stops logging. + yield b'2025-05-13T07:24:12.588Z ACTOR: going quiet\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(_silent_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() + + streaming_thread = streamed_log.start() + try: + # The buffered line proves the thread is inside the read loop; a fixed sleep could end before it gets there. + deadline = time.monotonic() + 5 + while not streamed_log._stream_buffer and time.monotonic() < deadline: + time.sleep(0.01) + assert streamed_log._stream_buffer, 'streaming thread never reached the blocking read' + + # Call stop() from a helper thread so the test cannot hang if the bound regresses. + stop_thread = threading.Thread(target=streamed_log.stop) + stop_thread.start() + stop_thread.join(timeout=5) + assert not stop_thread.is_alive(), 'stop() did not return within its bound on a silent stream' + finally: + release_server.set() + # `stop` leaves the thread running, so reap it instead of leaking it into the rest of the session. + streaming_thread.join(timeout=5) + + assert not streaming_thread.is_alive() + restarted_thread = streamed_log.start() + assert restarted_thread is not streaming_thread + streamed_log.stop() + restarted_thread.join(timeout=5) + + +def test_streamed_log_sync_stop_reports_failing_stream_close( + caplog: LogCaptureFixture, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A custom transport whose `close` raises is reported by `stop` instead of failing the caller.""" + monkeypatch.setattr(StreamedLog, '_stop_timeout_s', 0.1) + release_thread = threading.Event() + # Stand in for the streaming thread, so `stop` closes the test's response rather than racing for a real one. + monkeypatch.setattr(StreamedLog, '_stream_log', lambda _self: release_thread.wait(timeout=30)) + + logger = logging.getLogger('apify_client.tests.failing_stream_close') + streamed_log = StreamedLog(log_client=Mock(), to_logger=logger) + failing_stream = Mock() + failing_stream.close.side_effect = RuntimeError('close failed') + + streaming_thread = streamed_log.start() + try: + streamed_log._log_stream = failing_stream + + with caplog.at_level(logging.DEBUG, logger=logger.name): + streamed_log.stop() + + failing_stream.close.assert_called_once() + assert any('Closing the log stream failed' in record.message for record in caplog.records) + finally: + release_thread.set() + streaming_thread.join(timeout=5) + + def test_logger_once_logs_the_first_call(caplog: LogCaptureFixture) -> None: """Test the first call with a given key is logged.""" logger = logging.getLogger('apify_client.tests.log_once_first')