Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 20 additions & 17 deletions src/apify/_actor.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
)
Expand Down Expand Up @@ -1781,34 +1781,40 @@ 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'
self.log.info(f'[{title}]: {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(
Expand Down Expand Up @@ -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)

Expand Down
4 changes: 2 additions & 2 deletions src/apify/_consts.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""
Expand Down
68 changes: 68 additions & 0 deletions tests/unit/actor/test_actor_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
17 changes: 9 additions & 8 deletions tests/unit/actor/test_actor_lifecycle.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,8 @@
if TYPE_CHECKING:
from collections.abc import AsyncGenerator, Callable

from ..conftest import ApifyClientAsyncPatcher


@pytest.fixture(
params=[
Expand Down Expand Up @@ -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(
Expand All @@ -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:
Expand Down
Loading