From e42437a200e1eb71fe4af29c828e6a3b1c9d0102 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Fri, 16 Oct 2026 12:23:58 +0200 Subject: [PATCH 1/3] feat: forward crawler status messages to the run status message --- src/apify/_actor.py | 23 +++++++ tests/unit/actor/test_actor_lifecycle.py | 78 +++++++++++++++++++++++- 2 files changed, 100 insertions(+), 1 deletion(-) diff --git a/src/apify/_actor.py b/src/apify/_actor.py index 150cf924..6135f9e1 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, @@ -183,6 +184,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 @@ -245,6 +247,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 +267,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,6 +299,9 @@ 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: @@ -1818,6 +1828,19 @@ 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, so they cannot overwrite the terminal one. + """ + if self._is_exiting: + return + + try: + await self.set_status_message(event_data.message) + 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/tests/unit/actor/test_actor_lifecycle.py b/tests/unit/actor/test_actor_lifecycle.py index 19e4ea99..0748c4eb 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,82 @@ 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_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)] + + @pytest.mark.parametrize( ('first_with_call', 'second_with_call'), [ From a0b8a2c15362893401cbce70b017aef7c83c9406 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Fri, 16 Oct 2026 12:36:20 +0200 Subject: [PATCH 2/3] fix: serialize forwarded crawler status with the terminal status message --- src/apify/_actor.py | 18 ++++++++----- tests/unit/actor/test_actor_lifecycle.py | 32 ++++++++++++++++++++++++ 2 files changed, 44 insertions(+), 6 deletions(-) diff --git a/src/apify/_actor.py b/src/apify/_actor.py index 6135f9e1..761d7eb1 100644 --- a/src/apify/_actor.py +++ b/src/apify/_actor.py @@ -175,6 +175,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. @@ -212,6 +215,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 @@ -305,7 +309,8 @@ async def __aexit__( 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') @@ -1831,13 +1836,14 @@ async def use_state( 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, so they cannot overwrite the terminal one. + 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. """ - if self._is_exiting: - return - try: - await self.set_status_message(event_data.message) + async with self._status_message_lock: + if self._is_exiting: + return + await self.set_status_message(event_data.message) except Exception: self.log.warning('Failed to forward the crawler status message to the Actor run', exc_info=True) diff --git a/tests/unit/actor/test_actor_lifecycle.py b/tests/unit/actor/test_actor_lifecycle.py index 0748c4eb..39ee1133 100644 --- a/tests/unit/actor/test_actor_lifecycle.py +++ b/tests/unit/actor/test_actor_lifecycle.py @@ -547,6 +547,38 @@ async def test_crawler_status_does_not_follow_terminal_status_message(monkeypatc 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'), [ From f16a0c67b0d62e0d80059809eaf6dccb5f51cdd0 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Fri, 16 Oct 2026 12:59:35 +0200 Subject: [PATCH 3/3] feat: time out forwarded crawler status updates after 1 second --- src/apify/_actor.py | 16 +++++++++++--- src/apify/_consts.py | 3 +++ tests/unit/actor/test_actor_lifecycle.py | 27 ++++++++++++++++++++++++ 3 files changed, 43 insertions(+), 3 deletions(-) diff --git a/src/apify/_actor.py b/src/apify/_actor.py index 761d7eb1..54fdfd18 100644 --- a/src/apify/_actor.py +++ b/src/apify/_actor.py @@ -38,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 @@ -1837,13 +1843,17 @@ async def _forward_crawler_status(self, event_data: EventCrawlerStatusData) -> N """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. + 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 self.set_status_message(event_data.message) + 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) 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 39ee1133..65ac5fee 100644 --- a/tests/unit/actor/test_actor_lifecycle.py +++ b/tests/unit/actor/test_actor_lifecycle.py @@ -528,6 +528,33 @@ async def test_failed_crawler_status_forwarding_logs_warning( 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)