fix: Prevent losing a status message when redirecting Actor run logs

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)
 

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,