feat: forward crawler status messages to the run status message

src/apify/_actor.py

@@ -21,6 +21,7 @@
 from crawlee.events import (
     Event,
     EventAbortingData,
+    EventCrawlerStatusData,
     EventExitData,
     EventListener,
     EventMigratingData,
@@ -37,7 +38,13 @@
 )
 from apify._child_runs import ChildRunRegistry
 from apify._configuration import Configuration
-from apify._consts import EVENT_LISTENERS_TIMEOUT, EXIT_CODE_ERROR_USER_FUNCTION_THREW, ActorEnvVars, ApifyEnvVars
+from apify._consts import (
+    CRAWLER_STATUS_FORWARDING_TIMEOUT,
+    EVENT_LISTENERS_TIMEOUT,
+    EXIT_CODE_ERROR_USER_FUNCTION_THREW,
+    ActorEnvVars,
+    ApifyEnvVars,
+)
 from apify._crypto import decrypt_input_secrets, load_private_key
 from apify._proxy_configuration import ProxyConfiguration
 from apify._utils import docs_group, docs_name, ensure_context, get_system_info, is_running_in_ipython
@@ -174,6 +181,9 @@ def __init__(
         self._is_exiting = False
         """Whether the Actor is currently exiting."""
 
+        self._status_message_lock = asyncio.Lock()
+        """Serializes forwarded crawler status messages with the terminal status message."""
+
     async def __aenter__(self) -> Self:
         """Enter the Actor context.
 
@@ -183,6 +193,7 @@ async def __aenter__(self) -> Self:
         - Configures the event manager and starts periodic state persistence.
         - Initializes the charging manager for handling charging events.
         - Loads the named child runs recorded by an earlier attempt of this Actor run.
+        - On the Apify platform, forwards the status messages of Crawlee crawlers to the run status message.
         - Configures logging after all core services are registered.
 
         This method must be called exactly once per Actor instance. Re-initializing an Actor or having multiple
@@ -210,6 +221,7 @@ async def __aenter__(self) -> Self:
         # Update the global Actor proxy to refer to this instance.
         cast('Proxy', Actor).__wrapped__ = self  # ty: ignore[invalid-assignment]
         self._is_exiting = False
+        self._status_message_lock = asyncio.Lock()
 
         # Initialize the storage client and register it in the service locator.
         _ = self._storage_client
@@ -245,6 +257,10 @@ async def __aenter__(self) -> Self:
             finally:
                 await self.event_manager.__aexit__(None, None, None)
             raise
+
+        if self.is_at_home():
+            self.event_manager.on(event=Event.CRAWLER_STATUS, listener=self._forward_crawler_status)
+
         return self
 
     async def __aexit__(
@@ -261,6 +277,7 @@ async def __aexit__(
         so interrupt and cancellation semantics are preserved. Otherwise, the Actor exits cleanly. In every
         case the Actor:
 
+        - Stops forwarding crawler status messages to the run status message.
         - Cancels periodic `PERSIST_STATE` events.
         - Sends a final `PERSIST_STATE` event.
         - Waits for all event listeners to finish.
@@ -292,10 +309,14 @@ async def __aexit__(
         self._is_exiting = True
         self.log.info('Exiting Actor', extra={'exit_code': self.exit_code})
 
+        # Stop forwarding crawler status messages, so that none of them overwrites the terminal one.
+        self.event_manager.off(event=Event.CRAWLER_STATUS, listener=self._forward_crawler_status)
+
         async def finalize() -> None:
             if self.status_message is not None:
                 try:
-                    await self.set_status_message(self.status_message, is_terminal=True)
+                    async with self._status_message_lock:
+                        await self.set_status_message(self.status_message, is_terminal=True)
                 except Exception:
                     self.log.exception('Failed to set terminal status message')
 
@@ -1818,6 +1839,24 @@ async def use_state(
         kvs = await self.open_key_value_store(name=kvs_name)
         return await kvs.get_auto_saved_value(key or self._ACTOR_STATE_KEY, default_value)
 
+    async def _forward_crawler_status(self, event_data: EventCrawlerStatusData) -> None:
+        """Set a crawler status message as the run status message on the Apify platform.
+
+        Messages that arrive once the Actor is exiting are dropped, and the terminal message waits for an update that
+        is already in flight, so no forwarded message can overwrite the terminal one. Each update is bounded by
+        `CRAWLER_STATUS_FORWARDING_TIMEOUT`, so a slow or unavailable API can't hold up the crawler or the Actor exit.
+        """
+        try:
+            async with self._status_message_lock:
+                if self._is_exiting:
+                    return
+                await asyncio.wait_for(
+                    self.set_status_message(event_data.message),
+                    timeout=CRAWLER_STATUS_FORWARDING_TIMEOUT.total_seconds(),
+                )
+        except Exception:
+            self.log.warning('Failed to forward the crawler status message to the Actor run', exc_info=True)
+
     async def _save_actor_state(self) -> None:
         async def safe_persist(kvs_name: str | None) -> None:
             try:

src/apify/_consts.py

@@ -7,6 +7,9 @@
 EVENT_LISTENERS_TIMEOUT = timedelta(seconds=5)
 """Timeout for waiting on event listeners to finish during Actor exit."""
 
+CRAWLER_STATUS_FORWARDING_TIMEOUT = timedelta(seconds=1)
+"""Timeout for setting a forwarded crawler status message as the run status message."""
+
 BASE64_REGEXP = '[-A-Za-z0-9+/]*={0,3}'
 """Regex fragment matching a single base64-encoded component."""
 

tests/unit/actor/test_actor_lifecycle.py

@@ -15,7 +15,7 @@
 import websockets.asyncio.server
 
 from apify_client._models import Run
-from crawlee.events._types import Event, EventAbortingData, EventPersistStateData
+from crawlee.events._types import Event, EventAbortingData, EventCrawlerStatusData, EventPersistStateData
 
 from ..._utils import poll_until_condition
 from apify import Actor
@@ -471,6 +471,141 @@ async def test_failing_terminal_status_message_does_not_abort_exit(
     save_actor_state.assert_called_once()
 
 
+def set_up_platform_env(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv(ApifyEnvVars.IS_AT_HOME, '1')
+    monkeypatch.setenv(ActorEnvVars.RUN_ID, 'test-run-id')
+    # On the platform, init fetches the run for the charging manager and reads the child run registry from the API.
+    monkeypatch.setattr(ChargingManagerImplementation, '__aenter__', AsyncMock())
+    monkeypatch.setattr(ChargingManagerImplementation, '__aexit__', AsyncMock())
+    monkeypatch.setattr(ChildRunRegistry, 'load', AsyncMock(return_value={}))
+
+
+@pytest.mark.parametrize(
+    ('is_at_home', 'expected_calls'),
+    [
+        pytest.param(True, [mock.call('Crawled 1/2 pages')], id='on platform'),
+        pytest.param(False, [], id='locally'),
+    ],
+)
+async def test_crawler_status_is_forwarded_to_run_status_message(
+    monkeypatch: pytest.MonkeyPatch,
+    *,
+    is_at_home: bool,
+    expected_calls: list[Any],
+) -> None:
+    """A `CRAWLER_STATUS` event sets the run status message on the platform and does nothing locally."""
+    if is_at_home:
+        set_up_platform_env(monkeypatch)
+    set_status_message = AsyncMock()
+    monkeypatch.setattr(_ActorType, 'set_status_message', set_status_message)
+
+    async with Actor() as actor:
+        actor.event_manager.emit(
+            event=Event.CRAWLER_STATUS,
+            event_data=EventCrawlerStatusData(message='Crawled 1/2 pages', crawler_id=1),
+        )
+        await actor.event_manager.wait_for_all_listeners_to_complete()
+        assert set_status_message.call_args_list == expected_calls
+
+
+async def test_failed_crawler_status_forwarding_logs_warning(
+    caplog: pytest.LogCaptureFixture,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A failure to set a forwarded crawler status message is logged as a warning and does not propagate."""
+    set_up_platform_env(monkeypatch)
+    monkeypatch.setattr(_ActorType, 'set_status_message', AsyncMock(side_effect=RuntimeError('Update failed')))
+
+    async with Actor() as actor:
+        actor.event_manager.emit(
+            event=Event.CRAWLER_STATUS,
+            event_data=EventCrawlerStatusData(message='Crawled 1/2 pages', crawler_id=1),
+        )
+        await actor.event_manager.wait_for_all_listeners_to_complete()
+
+    records = [r for r in caplog.records if r.msg == 'Failed to forward the crawler status message to the Actor run']
+    assert len(records) == 1
+    assert records[0].levelno == logging.WARNING
+
+
+async def test_slow_crawler_status_forwarding_times_out(
+    caplog: pytest.LogCaptureFixture,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A forwarded crawler status update that exceeds the timeout is abandoned with a warning."""
+    set_up_platform_env(monkeypatch)
+    monkeypatch.setattr('apify._actor.CRAWLER_STATUS_FORWARDING_TIMEOUT', timedelta(milliseconds=10))
+
+    async def set_status_message(_self: Any, _status_message: str, *, is_terminal: bool | None = None) -> None:
+        if not is_terminal:
+            await asyncio.sleep(60)
+
+    monkeypatch.setattr(_ActorType, 'set_status_message', set_status_message)
+
+    async with Actor() as actor:
+        actor.event_manager.emit(
+            event=Event.CRAWLER_STATUS,
+            event_data=EventCrawlerStatusData(message='Crawled 1/2 pages', crawler_id=1),
+        )
+        await asyncio.wait_for(actor.event_manager.wait_for_all_listeners_to_complete(), timeout=5)
+
+    records = [r for r in caplog.records if r.msg == 'Failed to forward the crawler status message to the Actor run']
+    assert len(records) == 1
+    assert records[0].exc_info is not None
+    assert isinstance(records[0].exc_info[1], TimeoutError)
+
+
+async def test_crawler_status_does_not_follow_terminal_status_message(monkeypatch: pytest.MonkeyPatch) -> None:
+    """A crawler status message emitted right before exit does not overwrite the terminal status message."""
+    set_up_platform_env(monkeypatch)
+    set_status_message = AsyncMock()
+    monkeypatch.setattr(_ActorType, 'set_status_message', set_status_message)
+
+    actor = Actor()
+    await actor.init()
+
+    # The listener task only starts at the first `await` inside `exit()`, after the Actor began exiting.
+    actor.event_manager.emit(
+        event=Event.CRAWLER_STATUS,
+        event_data=EventCrawlerStatusData(message='Crawled 1/2 pages', crawler_id=1),
+    )
+    await actor.exit(status_message='Done')
+
+    assert set_status_message.call_args_list == [mock.call('Done', is_terminal=True)]
+
+
+async def test_terminal_status_message_waits_for_in_flight_crawler_status(monkeypatch: pytest.MonkeyPatch) -> None:
+    """The terminal status message is sent only after a crawler status update already in flight completes."""
+    set_up_platform_env(monkeypatch)
+    updates = Mock()
+    forwarding_started = asyncio.Event()
+
+    async def set_status_message(_self: Any, status_message: str, *, is_terminal: bool | None = None) -> None:
+        updates.start(status_message)
+        if not is_terminal:
+            forwarding_started.set()
+            await asyncio.sleep(0.05)
+        updates.end(status_message)
+
+    monkeypatch.setattr(_ActorType, 'set_status_message', set_status_message)
+
+    actor = Actor()
+    await actor.init()
+    actor.event_manager.emit(
+        event=Event.CRAWLER_STATUS,
+        event_data=EventCrawlerStatusData(message='Crawled 1/2 pages', crawler_id=1),
+    )
+    await forwarding_started.wait()
+    await actor.exit(status_message='Done')
+
+    assert updates.mock_calls == [
+        mock.call.start('Crawled 1/2 pages'),
+        mock.call.end('Crawled 1/2 pages'),
+        mock.call.start('Done'),
+        mock.call.end('Done'),
+    ]
+
+
 @pytest.mark.parametrize(
     ('first_with_call', 'second_with_call'),
     [