diff --git a/src/apify/_actor.py b/src/apify/_actor.py index 150cf924..54fdfd18 100644 --- a/src/apify/_actor.py +++ b/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: diff --git a/src/apify/_consts.py b/src/apify/_consts.py index 7251a2a2..1b6e459f 100644 --- a/src/apify/_consts.py +++ b/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.""" diff --git a/tests/unit/actor/test_actor_lifecycle.py b/tests/unit/actor/test_actor_lifecycle.py index 19e4ea99..65ac5fee 100644 --- a/tests/unit/actor/test_actor_lifecycle.py +++ b/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'), [