diff --git a/src/apify_client/_resource_clients/actor.py b/src/apify_client/_resource_clients/actor.py index 62170bbe..705a4c27 100644 --- a/src/apify_client/_resource_clients/actor.py +++ b/src/apify_client/_resource_clients/actor.py @@ -371,7 +371,13 @@ def call( if logger == 'default': logger = None - with run_client.get_status_message_watcher(to_logger=logger), run_client.get_streamed_log(to_logger=logger): + # With the default logger, each helper rebuilds the same redirect logger from scratch, so both must exist before + # either starts polling; otherwise the streamed log reconfigures a logger the status watcher thread is already + # writing into. + status_redirector = run_client.get_status_message_watcher(to_logger=logger) + streamed_log = run_client.get_streamed_log(to_logger=logger) + + with status_redirector, streamed_log: return run_client.wait_for_finish(wait_duration=wait_duration) def build( @@ -868,6 +874,9 @@ async def call( if logger == 'default': logger = None + # With the default logger, each helper rebuilds the same redirect logger from scratch, so both must exist before + # either starts polling; otherwise the streamed log reconfigures a logger the status watcher task is already + # writing into. status_redirector = await run_client.get_status_message_watcher(to_logger=logger) streamed_log = await run_client.get_streamed_log(to_logger=logger) diff --git a/tests/unit/test_logging.py b/tests/unit/test_logging.py index 35b10e2e..fb67b927 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 @@ -14,7 +15,8 @@ from apify_client import ApifyClient, ApifyClientAsync from apify_client._logging import LoggerOnce, RedirectLogFormatter -from apify_client._status_message_watcher import StatusMessageWatcherBase +from apify_client._resource_clients import run as run_module +from apify_client._status_message_watcher import StatusMessageWatcher, StatusMessageWatcherBase from apify_client._streamed_log import StreamedLog, StreamedLogAsync, StreamedLogBase if TYPE_CHECKING: @@ -24,6 +26,7 @@ from pytest_httpserver import HTTPServer from apify_client._literals import ActorJobStatus + from apify_client._models import Run from apify_client.http_clients import HttpClient, HttpClientAsync _MOCKED_RUN_ID = 'mocked_run_id' @@ -372,6 +375,63 @@ def test_actor_call_redirect_logs_to_default_logger_sync( ) +@pytest.mark.usefixtures('mock_api', 'propagate_stream_logs', 'reduce_final_timeout_for_status_message_redirector') +def test_actor_call_sync_does_not_reconfigure_logger_used_by_running_watcher( + caplog: LogCaptureFixture, + httpserver: HTTPServer, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """No status message is lost when `call` builds the streamed log while the status watcher already runs.""" + # The events pin the damaging interleaving instead of relying on thread scheduling: the watcher holds its first + # message until the logger has been rebuilt, and a rebuild that finds the watcher already running returns only + # after that message has been logged - i.e. before `StreamedLog.__init__` re-enables propagation. + watcher_started = threading.Event() + logger_rebuilt = threading.Event() + watcher_logged = threading.Event() + + original_start = StatusMessageWatcher.start + + def recording_start(self: StatusMessageWatcher) -> threading.Thread: + thread = original_start(self) + watcher_started.set() + return thread + + original_create_redirect_logger = run_module.create_redirect_logger + create_calls = itertools.count(1) + + def instrumented_create_redirect_logger(name: str) -> logging.Logger: + to_logger = original_create_redirect_logger(name) + if next(create_calls) >= 2: + logger_rebuilt.set() + if watcher_started.is_set(): + assert watcher_logged.wait(timeout=5) + return to_logger + + original_log_run_data = StatusMessageWatcherBase._log_run_data + + def gated_log_run_data(self: StatusMessageWatcherBase, run_data: Run | None) -> bool: + logger_rebuilt.wait(timeout=5) + more_data_expected = original_log_run_data(self, run_data) + watcher_logged.set() + return more_data_expected + + monkeypatch.setattr(StatusMessageWatcher, 'start', recording_start) + monkeypatch.setattr(run_module, 'create_redirect_logger', instrumented_create_redirect_logger) + monkeypatch.setattr(StatusMessageWatcherBase, '_log_run_data', gated_log_run_data) + + api_url = httpserver.url_for('/').removesuffix('/') + + logger_name = f'apify.{_MOCKED_ACTOR_NAME} runId:{_MOCKED_RUN_ID}' + actor_client = ApifyClient(token='mocked_token', api_url=api_url).actor(actor_id=_MOCKED_ACTOR_ID) + + with caplog.at_level(logging.DEBUG, logger=logger_name): + actor_client.call() + + assert ('Status: RUNNING, Message: Initial message', logging.INFO) in { + (record.message, record.levelno) for record in caplog.records + } + + @pytest.mark.usefixtures('mock_api', 'propagate_stream_logs') async def test_actor_call_no_redirect_logs_async( caplog: LogCaptureFixture,