From 57f1a394c7341eeea246edba4d5f5013850c6fe6 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Fri, 21 Aug 2026 13:37:37 +0200 Subject: [PATCH 1/9] fix: Keep a failing background watcher from breaking a successful actor call --- src/apify_client/_status_message_watcher.py | 41 ++++-- src/apify_client/_streamed_log.py | 34 ++++- tests/unit/test_logging.py | 148 ++++++++++++++++++++ 3 files changed, 207 insertions(+), 16 deletions(-) diff --git a/src/apify_client/_status_message_watcher.py b/src/apify_client/_status_message_watcher.py index a1bf1e38..d8a4026c 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 + # The poll cannot continue, so warn and let the task end instead of leaking a traceback. + self._to_logger.warning('Status message redirection stopped: the status request timed out.') + else: + # A failed poll must not escape into `stop`, where it would surface as a failure of the run itself. + 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 can never 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 + # The poll cannot continue, so warn and let the thread end instead of leaking 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; log it instead. + 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..ef095230 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 ends the read on a transport that honours it, but Impit's blocking read is not + interruptible, 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: @@ -127,13 +136,21 @@ 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, for up to `_stop_timeout_s`, for it to finish. + + A thread that outlives the wait keeps `_stop_logging` set, so it exits after at most one more chunk, and is + a daemon, so it cannot hold up interpreter shutdown. Its handle is retained until it ends, which keeps `start` + from reviving it alongside a second thread reading into 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 + if self._log_stream is not None: + # On a transport that honours it, this releases the connection and ends a read blocked on a silent stream. + self._log_stream.close() + self._streaming_thread.join(timeout=self._stop_timeout_s) + if not self._streaming_thread.is_alive(): + 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 +168,24 @@ 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 and end a read blocked on a silent stream. + self._log_stream = log_stream try: + # `stop` may have already run, back when there was no response 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: + # `stop` closed the response to end a blocked read, so any resulting error is expected. + 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/unit/test_logging.py b/tests/unit/test_logging.py index 35b10e2e..8af00383 100644 --- a/tests/unit/test_logging.py +++ b/tests/unit/test_logging.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import itertools import json import logging import threading @@ -1015,6 +1016,153 @@ def generate_logs() -> Iterator[bytes]: assert any('ACTOR: still running' in record.message for record in caplog.records) +_POLLS_BEFORE_FAILURE = 3 +"""Plain (non-`waitForFinish`) run-status requests served successfully before the endpoint starts rejecting them. + +`call` spends two on setup - one in `get_status_message_watcher`, one in `get_streamed_log` - and the watcher's first +poll takes the third, so every request beyond this count is a watcher poll. +""" + +_POLL_FAILURE_FINAL_SLEEP_S = 4 +"""Long enough for the watcher, which `call` polls once per second, to reach the rejecting endpoint before exiting.""" + + +@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) + plain_requests = itertools.count() + + def _status_handler(request: Request) -> Response: + if 'waitForFinish' in request.args: + # `wait_for_finish` keeps succeeding, so the run itself reads as a success. + return Response(response=json.dumps({'data': finished_run}), status=200, mimetype='application/json') + if next(plain_requests) >= _POLLS_BEFORE_FAILURE: + 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') + + # Registered before `_register_run_and_actor_endpoints` so this handler wins for the run endpoint - + # pytest-httpserver matches permanent handlers in registration order. + 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(_streaming_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 an uncaught exception 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 can never 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 stream that stays silent under the production `no_timeout`.""" + monkeypatch.setattr(StreamedLog, '_stop_timeout_s', 1) + + 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'' + 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: + # Give the streaming thread time to start and block inside iter_bytes. + time.sleep(0.3) + + # Call stop() from a helper thread so the test cannot hang indefinitely 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 here instead of leaking it into the rest of the session. + 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') From 54fbce4d55588f5857c1c694fe72048c49c50dde Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Fri, 21 Aug 2026 18:33:03 +0200 Subject: [PATCH 2/9] fix: Read the log stream handle once in StreamedLog.stop --- src/apify_client/_streamed_log.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/src/apify_client/_streamed_log.py b/src/apify_client/_streamed_log.py index ef095230..6e9a7b5b 100644 --- a/src/apify_client/_streamed_log.py +++ b/src/apify_client/_streamed_log.py @@ -145,9 +145,11 @@ def stop(self) -> None: if not self._streaming_thread: raise RuntimeError('Streaming thread is not active') self._stop_logging = True - if self._log_stream is not None: + # Read once, because the streaming thread clears the attribute as soon as the stream ends. + log_stream = self._log_stream + if log_stream is not None: # On a transport that honours it, this releases the connection and ends a read blocked on a silent stream. - self._log_stream.close() + log_stream.close() self._streaming_thread.join(timeout=self._stop_timeout_s) if not self._streaming_thread.is_alive(): self._streaming_thread = None From 845fa0c6e502a7853f9e5f2e96878cd75d149761 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Fri, 21 Aug 2026 18:33:37 +0200 Subject: [PATCH 3/9] fix: Let StreamedLog.start recover from a stop that outlived its timeout --- src/apify_client/_streamed_log.py | 4 ++-- tests/unit/test_logging.py | 8 +++++++- 2 files changed, 9 insertions(+), 3 deletions(-) diff --git a/src/apify_client/_streamed_log.py b/src/apify_client/_streamed_log.py index 6e9a7b5b..e2f294bb 100644 --- a/src/apify_client/_streamed_log.py +++ b/src/apify_client/_streamed_log.py @@ -127,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. @@ -139,7 +139,7 @@ def stop(self) -> None: """Signal the streaming thread to stop logging and wait, for up to `_stop_timeout_s`, for it to finish. A thread that outlives the wait keeps `_stop_logging` set, so it exits after at most one more chunk, and is - a daemon, so it cannot hold up interpreter shutdown. Its handle is retained until it ends, which keeps `start` + a daemon, so it cannot hold up interpreter shutdown. Its handle is kept while it is alive, which keeps `start` from reviving it alongside a second thread reading into the same buffer. """ if not self._streaming_thread: diff --git a/tests/unit/test_logging.py b/tests/unit/test_logging.py index 8af00383..0ab06b59 100644 --- a/tests/unit/test_logging.py +++ b/tests/unit/test_logging.py @@ -1124,7 +1124,7 @@ def test_streamed_log_sync_stop_returns_on_silent_stream( httpserver: HTTPServer, monkeypatch: pytest.MonkeyPatch, ) -> None: - """`stop` returns within its bound on a stream that stays silent under the production `no_timeout`.""" + """`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() @@ -1162,6 +1162,12 @@ def generate_logs() -> Iterator[bytes]: # `stop` leaves the thread running, so reap it here 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_logger_once_logs_the_first_call(caplog: LogCaptureFixture) -> None: """Test the first call with a given key is logged.""" From e1d688c9312767a1faea004d17502f1d5cebbe7a Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Fri, 21 Aug 2026 18:34:19 +0200 Subject: [PATCH 4/9] fix: Report StreamedLog failures during stop instead of swallowing them --- src/apify_client/_streamed_log.py | 20 ++++++++++++++------ tests/unit/test_logging.py | 31 ++++++++++++++++++++++++++++++- 2 files changed, 44 insertions(+), 7 deletions(-) diff --git a/src/apify_client/_streamed_log.py b/src/apify_client/_streamed_log.py index e2f294bb..a0a16ef5 100644 --- a/src/apify_client/_streamed_log.py +++ b/src/apify_client/_streamed_log.py @@ -148,10 +148,16 @@ def stop(self) -> None: # Read once, because the streaming thread clears the attribute as soon as the stream ends. log_stream = self._log_stream if log_stream is not None: - # On a transport that honours it, this releases the connection and ends a read blocked on a silent stream. - log_stream.close() + try: + log_stream.close() + except Exception: + # A custom transport whose `close` fails must not turn a finished run into a failed call. + self._to_logger.exception('Closing the log stream failed:') self._streaming_thread.join(timeout=self._stop_timeout_s) - if not self._streaming_thread.is_alive(): + if self._streaming_thread.is_alive(): + # Without this, log messages arriving after `stop` returned have no visible 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: @@ -170,10 +176,10 @@ 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 and end a read blocked on a silent stream. + # Published so `stop` can close the response. self._log_stream = log_stream try: - # `stop` may have already run, back when there was no response for it to close. + # `stop` may have run before the response existed for it to close. if self._stop_logging: return for data in log_stream.iter_bytes(): @@ -186,7 +192,9 @@ def _stream_log(self) -> None: self._log_buffer_content(include_last_part=True) except Exception as exc: if self._stop_logging: - # `stop` closed the response to end a blocked read, so any resulting error is expected. + # A stop is in progress, so the failure is expected. Report it quietly rather than not at all, since + # this also catches a flush of the buffered tail that failed for a reason of its own. + 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. diff --git a/tests/unit/test_logging.py b/tests/unit/test_logging.py index 0ab06b59..427d3703 100644 --- a/tests/unit/test_logging.py +++ b/tests/unit/test_logging.py @@ -8,7 +8,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 @@ -1169,6 +1169,35 @@ def generate_logs() -> Iterator[bytes]: 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 the response `stop` closes is the test's and not one it has to race for. + 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') From de0a7a496f405f2a84276d9162abf66c5e9f2aec Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Fri, 21 Aug 2026 18:35:00 +0200 Subject: [PATCH 5/9] test: Gate the failing status poll on the log stream request instead of a request count --- tests/unit/test_logging.py | 23 +++++++++++------------ 1 file changed, 11 insertions(+), 12 deletions(-) diff --git a/tests/unit/test_logging.py b/tests/unit/test_logging.py index 427d3703..d7c91b06 100644 --- a/tests/unit/test_logging.py +++ b/tests/unit/test_logging.py @@ -1,7 +1,6 @@ from __future__ import annotations import asyncio -import itertools import json import logging import threading @@ -1016,15 +1015,8 @@ def generate_logs() -> Iterator[bytes]: assert any('ACTOR: still running' in record.message for record in caplog.records) -_POLLS_BEFORE_FAILURE = 3 -"""Plain (non-`waitForFinish`) run-status requests served successfully before the endpoint starts rejecting them. - -`call` spends two on setup - one in `get_status_message_watcher`, one in `get_streamed_log` - and the watcher's first -poll takes the third, so every request beyond this count is a watcher poll. -""" - _POLL_FAILURE_FINAL_SLEEP_S = 4 -"""Long enough for the watcher, which `call` polls once per second, to reach the rejecting endpoint before exiting.""" +"""Long enough for the watcher, which polls once per second under `call`'s defaults, to reach the rejecting endpoint.""" @pytest.fixture @@ -1033,13 +1025,16 @@ def mock_api_failing_status_poll(httpserver: HTTPServer) -> None: 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) - plain_requests = itertools.count() + # `call` asks for the log stream only once both of its own status requests have returned, so that request marks + # the point from which every plain status request is a watcher poll. A request count cannot tell the two apart, + # because the watcher thread starts polling before the second setup request is issued. + setup_done = threading.Event() def _status_handler(request: Request) -> Response: if 'waitForFinish' in request.args: # `wait_for_finish` keeps succeeding, so the run itself reads as a success. return Response(response=json.dumps({'data': finished_run}), status=200, mimetype='application/json') - if next(plain_requests) >= _POLLS_BEFORE_FAILURE: + if setup_done.is_set(): return Response( response=json.dumps({'error': {'type': 'insufficient-permissions', 'message': 'Poll rejected'}}), status=403, @@ -1047,6 +1042,10 @@ def _status_handler(request: Request) -> Response: ) 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` so this handler wins for the run endpoint - # pytest-httpserver matches permanent handlers in registration order. httpserver.expect_request(f'/v2/actor-runs/{_MOCKED_RUN_ID}', method='GET').respond_with_handler(_status_handler) @@ -1056,7 +1055,7 @@ def _status_handler(request: Request) -> Response: ) httpserver.expect_request( f'/v2/actor-runs/{_MOCKED_RUN_ID}/log', method='GET', query_string='stream=true&raw=true' - ).respond_with_handler(_streaming_log_handler) + ).respond_with_handler(_log_handler) @pytest.mark.usefixtures('mock_api_failing_status_poll', 'propagate_stream_logs') From 7b0be2404e35e746e959a480ea06e42a85afe6dd Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Fri, 21 Aug 2026 18:35:21 +0200 Subject: [PATCH 6/9] test: Isolate the finite stream timeout test from stop's own join bound --- tests/unit/test_logging.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/tests/unit/test_logging.py b/tests/unit/test_logging.py index d7c91b06..6bd41b98 100644 --- a/tests/unit/test_logging.py +++ b/tests/unit/test_logging.py @@ -835,6 +835,9 @@ 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. Left at its default it + # equals the `join` budget below, and the assertion can no longer tell a working stream timeout from a broken one. + monkeypatch.setattr(StreamedLog, '_stop_timeout_s', 30) release_server = threading.Event() From 069310b07023042135af07f73ee060e922cd6406 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Fri, 21 Aug 2026 18:35:52 +0200 Subject: [PATCH 7/9] test: Wait for the streamed log to reach its blocking read instead of sleeping --- tests/unit/test_logging.py | 28 ++++++++++++++++++---------- 1 file changed, 18 insertions(+), 10 deletions(-) diff --git a/tests/unit/test_logging.py b/tests/unit/test_logging.py index 6bd41b98..bfe77028 100644 --- a/tests/unit/test_logging.py +++ b/tests/unit/test_logging.py @@ -843,9 +843,9 @@ def test_streamed_log_sync_stop_unblocks_on_finite_stream_timeout( 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, so the test can wait for proof that the reader reached the read that then blocks + # (as a running Actor that stops logging would), rather than guessing at how long that takes. + 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') @@ -861,8 +861,12 @@ 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 instead would, if it ever ended + # too early, let the thread leave on the `_stop_logging` check without the stream timeout ever mattering. + 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) @@ -1133,9 +1137,9 @@ def test_streamed_log_sync_stop_returns_on_silent_stream( 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, so the test can wait for proof that the reader reached the read that then blocks + # (as a running Actor that stops logging would), rather than guessing at how long that takes. + 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') @@ -1151,8 +1155,12 @@ def generate_logs() -> Iterator[bytes]: streaming_thread = 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 instead would, if it ever ended + # too early, let the thread leave on the `_stop_logging` check and pass the test without exercising the bound. + 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 bound regresses. stop_thread = threading.Thread(target=streamed_log.stop) From b6d370a96d3058fbf7acef3f2bf991f1f1efc04a Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Fri, 21 Aug 2026 18:43:20 +0200 Subject: [PATCH 8/9] docs: Tighten the comments and docstrings around the watcher and streamed log --- src/apify_client/_status_message_watcher.py | 10 +++--- src/apify_client/_streamed_log.py | 20 +++++------ tests/unit/test_logging.py | 37 +++++++++------------ 3 files changed, 29 insertions(+), 38 deletions(-) diff --git a/src/apify_client/_status_message_watcher.py b/src/apify_client/_status_message_watcher.py index d8a4026c..030db953 100644 --- a/src/apify_client/_status_message_watcher.py +++ b/src/apify_client/_status_message_watcher.py @@ -128,10 +128,10 @@ async def _log_changed_status_message(self) -> None: await asyncio.sleep(self._check_period) except Exception as exc: if self._run_client._http_client.is_timeout_error(exc): # noqa: SLF001 - # The poll cannot continue, so warn and let the task end instead of leaking a traceback. + # 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`, where it would surface as a failure of the run itself. + # 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:') @@ -170,7 +170,7 @@ def start(self) -> Thread: if self._logging_thread: raise RuntimeError('Logging thread already active') self._stop_logging = False - # A daemon thread so a watcher still polling can never hold up interpreter shutdown. + # 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 @@ -207,8 +207,8 @@ def _log_changed_status_message(self) -> None: time.sleep(self._check_period) except Exception as exc: if self._run_client._http_client.is_timeout_error(exc): # noqa: SLF001 - # The poll cannot continue, so warn and let the thread end instead of leaking a traceback. + # 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; log it instead. + # 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 a0a16ef5..a9bb8c5a 100644 --- a/src/apify_client/_streamed_log.py +++ b/src/apify_client/_streamed_log.py @@ -102,8 +102,8 @@ class StreamedLog(StreamedLogBase): _stop_timeout_s: ClassVar[float] = 5 """Upper bound on how long `stop` waits for the streaming thread to finish. - Closing the response ends the read on a transport that honours it, but Impit's blocking read is not - interruptible, so without a bound `stop` would wait for the next chunk, which on a quiet run may be hours away. + 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: @@ -136,26 +136,25 @@ def start(self) -> Thread: return self._streaming_thread def stop(self) -> None: - """Signal the streaming thread to stop logging and wait, for up to `_stop_timeout_s`, 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 keeps `_stop_logging` set, so it exits after at most one more chunk, and is - a daemon, so it cannot hold up interpreter shutdown. Its handle is kept while it is alive, which keeps `start` - from reviving it alongside a second thread reading into the same buffer. + 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 - # Read once, because the streaming thread clears the attribute as soon as the stream ends. + # 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 custom transport whose `close` fails must not turn a finished run into a failed call. + # 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(): - # Without this, log messages arriving after `stop` returned have no visible explanation. + # 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 @@ -192,8 +191,7 @@ def _stream_log(self) -> None: self._log_buffer_content(include_last_part=True) except Exception as exc: if self._stop_logging: - # A stop is in progress, so the failure is expected. Report it quietly rather than not at all, since - # this also catches a flush of the buffered tail that failed for a reason of its own. + # 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 diff --git a/tests/unit/test_logging.py b/tests/unit/test_logging.py index bfe77028..03926b16 100644 --- a/tests/unit/test_logging.py +++ b/tests/unit/test_logging.py @@ -835,16 +835,14 @@ 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. Left at its default it - # equals the `join` budget below, and the assertion can no longer tell a working stream timeout from a broken one. + # 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]: - # One complete line, so the test can wait for proof that the reader reached the read that then blocks - # (as a running Actor that stops logging would), rather than guessing at how long that takes. + # 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) @@ -861,8 +859,7 @@ def generate_logs() -> Iterator[bytes]: streamed_log.start() try: - # The buffered line proves the thread is inside the read loop. A fixed sleep instead would, if it ever ended - # too early, let the thread leave on the `_stop_logging` check without the stream timeout ever mattering. + # 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) @@ -1023,7 +1020,7 @@ def generate_logs() -> Iterator[bytes]: _POLL_FAILURE_FINAL_SLEEP_S = 4 -"""Long enough for the watcher, which polls once per second under `call`'s defaults, to reach the rejecting endpoint.""" +"""Long enough for the once-per-second watcher poll to reach the rejecting endpoint.""" @pytest.fixture @@ -1032,14 +1029,13 @@ def mock_api_failing_status_poll(httpserver: HTTPServer) -> None: 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` asks for the log stream only once both of its own status requests have returned, so that request marks - # the point from which every plain status request is a watcher poll. A request count cannot tell the two apart, - # because the watcher thread starts polling before the second setup request is issued. + # `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 itself reads as a success. + # `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( @@ -1053,8 +1049,7 @@ def _log_handler(request: Request) -> Response: setup_done.set() return _streaming_log_handler(request) - # Registered before `_register_run_and_actor_endpoints` so this handler wins for the run endpoint - - # pytest-httpserver matches permanent handlers in registration order. + # 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( @@ -1092,7 +1087,7 @@ def test_actor_call_returns_run_when_status_poll_fails_sync( httpserver: HTTPServer, monkeypatch: pytest.MonkeyPatch, ) -> None: - """A failing status poll is logged by the watcher thread instead of leaking an uncaught exception out of it.""" + """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] = [] @@ -1114,7 +1109,7 @@ def test_actor_call_returns_run_when_status_poll_fails_sync( @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 can never hold up interpreter shutdown.""" + """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)) @@ -1137,8 +1132,7 @@ def test_streamed_log_sync_stop_returns_on_silent_stream( def _silent_handler(_request: Request) -> Response: def generate_logs() -> Iterator[bytes]: - # One complete line, so the test can wait for proof that the reader reached the read that then blocks - # (as a running Actor that stops logging would), rather than guessing at how long that takes. + # 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) @@ -1155,21 +1149,20 @@ def generate_logs() -> Iterator[bytes]: streaming_thread = streamed_log.start() try: - # The buffered line proves the thread is inside the read loop. A fixed sleep instead would, if it ever ended - # too early, let the thread leave on the `_stop_logging` check and pass the test without exercising the bound. + # 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 bound regresses. + # 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 here instead of leaking it into the rest of the session. + # `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() @@ -1186,7 +1179,7 @@ def test_streamed_log_sync_stop_reports_failing_stream_close( """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 the response `stop` closes is the test's and not one it has to race for. + # 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') From 46d720272eb638cf10b2ec1fbfd3256a0ad49a9b Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Fri, 21 Aug 2026 19:37:54 +0200 Subject: [PATCH 9/9] test: Cover a failing status poll during actor.call against the real API --- tests/integration/test_log.py | 54 +++++++++++++++++++++++++++++++++++ 1 file changed, 54 insertions(+) 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())