diff --git a/src/apify/_actor.py b/src/apify/_actor.py index 7bb01019..e3c0d323 100644 --- a/src/apify/_actor.py +++ b/src/apify/_actor.py @@ -40,9 +40,9 @@ from apify._child_runs import ChildRunRegistry from apify._configuration import Configuration from apify._consts import ( - CRAWLER_STATUS_FORWARDING_TIMEOUT, EVENT_LISTENERS_TIMEOUT, EXIT_CODE_ERROR_USER_FUNCTION_THREW, + STATUS_MESSAGE_UPDATE_TIMEOUT, ActorEnvVars, ApifyEnvVars, ) @@ -1781,12 +1781,15 @@ async def set_status_message( ) -> Run | None: """Set the status message for the current Actor run. + The status message is cosmetic, so a failed or timed-out update is logged as a warning and never raises. + Args: status_message: The status message to set to the run. is_terminal: Set this flag to True if this is the final status message of the Actor run. Returns: - The updated Actor run object. + The updated Actor run object, or `None` when no run was updated: when running locally, or when the update + failed or timed out. """ if not self.is_at_home(): title = 'Terminal status message' if is_terminal else 'Status message' @@ -1794,21 +1797,24 @@ async def set_status_message( return None # If is_at_home() is True, config.actor_run_id is always set - if not self.configuration.actor_run_id: + run_id = self.configuration.actor_run_id + if not run_id: raise RuntimeError('actor_run_id cannot be None when running on the Apify platform.') - run_client = self.apify_client.run(self.configuration.actor_run_id) - run = await run_client.update( - status_message=status_message, - is_status_message_terminal=is_terminal, - ) + run_client = self.apify_client.run(run_id) + update = run_client.update(status_message=status_message, is_status_message_terminal=is_terminal) - if run is None: - raise RuntimeError( - f'Failed to set status message for Actor run with ID "{self.configuration.actor_run_id}".' + try: + return await asyncio.wait_for(update, STATUS_MESSAGE_UPDATE_TIMEOUT.total_seconds()) + except TimeoutError: + self.log.warning( + f'Failed to set status message for Actor run with ID "{run_id}": ' + f'the update timed out after {STATUS_MESSAGE_UPDATE_TIMEOUT.total_seconds():g}s.' ) + except Exception as exc: + self.log.warning(f'Failed to set status message for Actor run with ID "{run_id}": {exc}') - return run + return None @_ensure_context async def create_proxy_configuration( @@ -1902,16 +1908,13 @@ async def _forward_crawler_status(self, event_data: EventCrawlerStatusData) -> N 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. + `STATUS_MESSAGE_UPDATE_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(), - ) + 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/src/apify/_consts.py b/src/apify/_consts.py index 1b6e459f..e7c39dea 100644 --- a/src/apify/_consts.py +++ b/src/apify/_consts.py @@ -7,8 +7,8 @@ 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.""" +STATUS_MESSAGE_UPDATE_TIMEOUT = timedelta(seconds=1) +"""How long `Actor.set_status_message()` waits for the run update before giving up with a warning.""" BASE64_REGEXP = '[-A-Za-z0-9+/]*={0,3}' """Regex fragment matching a single base64-encoded component.""" diff --git a/tests/unit/actor/test_actor_helpers.py b/tests/unit/actor/test_actor_helpers.py index 5ac9777a..e8b0d8f9 100644 --- a/tests/unit/actor/test_actor_helpers.py +++ b/tests/unit/actor/test_actor_helpers.py @@ -357,6 +357,74 @@ async def test_set_status_message_locally(caplog: pytest.LogCaptureFixture) -> N assert '[Status message]: test-status-message' in matching_records[0].message +async def test_set_status_message_returns_updated_run( + apify_client_async_patcher: ApifyClientAsyncPatcher, + fake_actor_run: Run, +) -> None: + """`Actor.set_status_message` on the platform updates the run and returns it.""" + apify_client_async_patcher.patch('run', 'update', return_value=fake_actor_run) + + async with Actor: + Actor.configuration.is_at_home = True + Actor.configuration.actor_run_id = 'some-run-id' + run = await Actor.set_status_message('almost there', is_terminal=True) + + assert run is fake_actor_run + update_kwargs = apify_client_async_patcher.calls['run']['update'][0][1] + assert update_kwargs == {'status_message': 'almost there', 'is_status_message_terminal': True} + + +async def test_set_status_message_warns_when_update_fails( + apify_client_async_patcher: ApifyClientAsyncPatcher, + caplog: pytest.LogCaptureFixture, +) -> None: + """`Actor.set_status_message` logs a warning and returns `None` when the run update raises.""" + + def failing_update(*_args: object, **_kwargs: object) -> Run: + raise RuntimeError('Boom') + + apify_client_async_patcher.patch('run', 'update', replacement_method=failing_update) + + async with Actor: + Actor.configuration.is_at_home = True + Actor.configuration.actor_run_id = 'some-run-id' + run = await Actor.set_status_message('almost there') + + assert run is None + matching_records = [record for record in caplog.records if 'Failed to set status message' in record.message] + assert len(matching_records) == 1 + assert matching_records[0].levelname == 'WARNING' + assert matching_records[0].message == 'Failed to set status message for Actor run with ID "some-run-id": Boom' + + +async def test_set_status_message_warns_when_update_times_out( + apify_client_async_patcher: ApifyClientAsyncPatcher, + caplog: pytest.LogCaptureFixture, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """`Actor.set_status_message` logs a warning and returns `None` when the run update does not finish in time.""" + + async def hanging_update(*_args: object, **_kwargs: object) -> Run: + await asyncio.Event().wait() + raise AssertionError('unreachable') + + apify_client_async_patcher.patch('run', 'update', replacement_method=hanging_update) + monkeypatch.setattr('apify._actor.STATUS_MESSAGE_UPDATE_TIMEOUT', timedelta(milliseconds=10)) + + async with Actor: + Actor.configuration.is_at_home = True + Actor.configuration.actor_run_id = 'some-run-id' + run = await Actor.set_status_message('almost there') + + assert run is None + matching_records = [record for record in caplog.records if 'Failed to set status message' in record.message] + assert len(matching_records) == 1 + assert matching_records[0].levelname == 'WARNING' + assert matching_records[0].message == ( + 'Failed to set status message for Actor run with ID "some-run-id": the update timed out after 0.01s.' + ) + + async def test_set_terminal_status_message_locally(caplog: pytest.LogCaptureFixture) -> None: caplog.set_level('INFO') async with Actor: diff --git a/tests/unit/actor/test_actor_lifecycle.py b/tests/unit/actor/test_actor_lifecycle.py index 86c02602..025747c1 100644 --- a/tests/unit/actor/test_actor_lifecycle.py +++ b/tests/unit/actor/test_actor_lifecycle.py @@ -27,6 +27,8 @@ if TYPE_CHECKING: from collections.abc import AsyncGenerator, Callable + from ..conftest import ApifyClientAsyncPatcher + @pytest.fixture( params=[ @@ -529,18 +531,18 @@ async def test_failed_crawler_status_forwarding_logs_warning( async def test_slow_crawler_status_forwarding_times_out( + apify_client_async_patcher: ApifyClientAsyncPatcher, 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)) + monkeypatch.setattr('apify._actor.STATUS_MESSAGE_UPDATE_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) + async def hanging_update(*_args: object, **_kwargs: object) -> None: + await asyncio.sleep(60) - monkeypatch.setattr(_ActorType, 'set_status_message', set_status_message) + apify_client_async_patcher.patch('run', 'update', replacement_method=hanging_update) async with Actor() as actor: actor.event_manager.emit( @@ -549,10 +551,9 @@ async def set_status_message(_self: Any, _status_message: str, *, is_terminal: b ) 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'] + records = [r for r in caplog.records if r.msg.startswith('Failed to set status message')] assert len(records) == 1 - assert isinstance(records[0].exc_info, tuple) - assert isinstance(records[0].exc_info[1], TimeoutError) + assert records[0].levelno == logging.WARNING async def test_crawler_status_does_not_follow_terminal_status_message(monkeypatch: pytest.MonkeyPatch) -> None: