From 20f5bbc188f4874c257b5f74534bee24aed65da1 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Tue, 25 Aug 2026 15:52:22 +0200 Subject: [PATCH 1/3] fix: Prevent losing a status message when redirecting Actor run logs --- src/apify_client/_resource_clients/actor.py | 8 ++- tests/unit/test_logging.py | 64 ++++++++++++++++++++- 2 files changed, 70 insertions(+), 2 deletions(-) diff --git a/src/apify_client/_resource_clients/actor.py b/src/apify_client/_resource_clients/actor.py index 62170bbe..0545ba22 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): + # Both helpers redirect into the same named logger and each rebuilds it from scratch, so they have to be + # constructed before either one starts polling - otherwise the streamed log reconfigures the logger that the + # status watcher thread is already writing into, and a status message emitted right then is lost. + 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( diff --git a/tests/unit/test_logging.py b/tests/unit/test_logging.py index 35b10e2e..f95e2dd0 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,65 @@ 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. + + Both helpers redirect into the same named logger, and building either one reconfigures that logger from scratch. + The events below pin the damaging interleaving so the outcome does not depend on thread scheduling: the watcher + holds its first message until the logger has just been rebuilt, and a rebuild that finds the watcher already + running keeps the half-configured logger in place until that message has been emitted into it.""" + 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(): + 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, From 1138a3ff86c63c0ef7cd5e587bc8d9fcdac47af6 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Tue, 25 Aug 2026 16:27:21 +0200 Subject: [PATCH 2/3] style: Tighten the redirect-logger comment and the regression test docstring --- src/apify_client/_resource_clients/actor.py | 5 ++--- tests/unit/test_logging.py | 10 ++++------ 2 files changed, 6 insertions(+), 9 deletions(-) diff --git a/src/apify_client/_resource_clients/actor.py b/src/apify_client/_resource_clients/actor.py index 0545ba22..04ccef44 100644 --- a/src/apify_client/_resource_clients/actor.py +++ b/src/apify_client/_resource_clients/actor.py @@ -371,9 +371,8 @@ def call( if logger == 'default': logger = None - # Both helpers redirect into the same named logger and each rebuilds it from scratch, so they have to be - # constructed before either one starts polling - otherwise the streamed log reconfigures the logger that the - # status watcher thread is already writing into, and a status message emitted right then is lost. + # Each helper rebuilds the shared redirect logger, 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) diff --git a/tests/unit/test_logging.py b/tests/unit/test_logging.py index f95e2dd0..9cd3e2e7 100644 --- a/tests/unit/test_logging.py +++ b/tests/unit/test_logging.py @@ -381,12 +381,10 @@ def test_actor_call_sync_does_not_reconfigure_logger_used_by_running_watcher( httpserver: HTTPServer, monkeypatch: pytest.MonkeyPatch, ) -> None: - """No status message is lost when `call` builds the streamed log while the status watcher already runs. - - Both helpers redirect into the same named logger, and building either one reconfigures that logger from scratch. - The events below pin the damaging interleaving so the outcome does not depend on thread scheduling: the watcher - holds its first message until the logger has just been rebuilt, and a rebuild that finds the watcher already - running keeps the half-configured logger in place until that message has been emitted into it.""" + """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() From 93a3f45aa864aa16e0a31af10876099fd00900dc Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Tue, 25 Aug 2026 20:35:23 +0200 Subject: [PATCH 3/3] test: Assert the redirect-logger handshake and mirror the ordering comment on the async call --- src/apify_client/_resource_clients/actor.py | 8 ++++++-- tests/unit/test_logging.py | 2 +- 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/src/apify_client/_resource_clients/actor.py b/src/apify_client/_resource_clients/actor.py index 04ccef44..705a4c27 100644 --- a/src/apify_client/_resource_clients/actor.py +++ b/src/apify_client/_resource_clients/actor.py @@ -371,8 +371,9 @@ def call( if logger == 'default': logger = None - # Each helper rebuilds the shared redirect logger, so both must exist before either starts polling; otherwise - # the streamed log reconfigures a logger the status watcher thread is already writing into. + # 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) @@ -873,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 9cd3e2e7..fb67b927 100644 --- a/tests/unit/test_logging.py +++ b/tests/unit/test_logging.py @@ -404,7 +404,7 @@ def instrumented_create_redirect_logger(name: str) -> logging.Logger: if next(create_calls) >= 2: logger_rebuilt.set() if watcher_started.is_set(): - watcher_logged.wait(timeout=5) + assert watcher_logged.wait(timeout=5) return to_logger original_log_run_data = StatusMessageWatcherBase._log_run_data