diff --git a/src/apify/_actor.py b/src/apify/_actor.py index c75cbf26..3db81bbc 100644 --- a/src/apify/_actor.py +++ b/src/apify/_actor.py @@ -4,6 +4,7 @@ import math import sys import warnings +from contextlib import contextmanager from dataclasses import asdict from datetime import UTC, datetime, timedelta from functools import cached_property @@ -41,7 +42,7 @@ if TYPE_CHECKING: import logging - from collections.abc import Callable, MutableMapping + from collections.abc import Callable, Iterator, MutableMapping from decimal import Decimal from types import TracebackType from typing import Self @@ -283,12 +284,15 @@ async def finalize() -> None: except Exception: self.log.exception('Failed to save Actor state') - try: - await asyncio.wait_for(finalize(), self._cleanup_timeout.total_seconds()) - except TimeoutError: - self.log.exception('Actor cleanup timed out') - finally: - self._active = False + # `exit()` / `fail()` may be called from within an event listener; detach that listener's own task for + # the duration of the cleanup so the waits below don't deadlock on it (see `_detach_current_listener_task`). + with self._detach_current_listener_task(): + try: + await asyncio.wait_for(finalize(), self._cleanup_timeout.total_seconds()) + except TimeoutError: + self.log.exception('Actor cleanup timed out') + finally: + self._active = False if reraise_control_flow: # Return without `sys.exit()` so the original exception re-raises. @@ -1488,6 +1492,28 @@ def _get_remaining_time(self) -> timedelta | None: ) return None + @contextmanager + def _detach_current_listener_task(self) -> Iterator[None]: + """Temporarily remove the current task from the event manager's listener-task set. + + If `exit()` / `fail()` runs inside an event listener, the current task is that listener's own tracked + task, so the cleanup waits below would deadlock on it and raise `RecursionError` on the timeout + cancellation. Detaching it skips it in those waits; restoring it lets the listener wrapper deregister it. + + Only a direct call on the listener's task is handled, not one from a task the listener itself spawns + (asyncio exposes no task ancestry). + """ + listener_tasks = self.event_manager._listener_tasks # noqa: SLF001 + current_task = asyncio.current_task() + is_listener_task = current_task is not None and current_task in listener_tasks + if is_listener_task: + listener_tasks.discard(current_task) + try: + yield + finally: + if is_listener_task: + listener_tasks.add(current_task) + Actor = cast('_ActorType', Proxy(_ActorType)) """The entry point of the SDK, through which all the Actor operations should be done.""" diff --git a/tests/unit/actor/test_actor_lifecycle.py b/tests/unit/actor/test_actor_lifecycle.py index a702cd09..0324e6fa 100644 --- a/tests/unit/actor/test_actor_lifecycle.py +++ b/tests/unit/actor/test_actor_lifecycle.py @@ -4,7 +4,7 @@ import contextlib import json import logging -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta from typing import TYPE_CHECKING, Any from unittest import mock from unittest.mock import AsyncMock, Mock @@ -14,7 +14,7 @@ import websockets.asyncio.server from apify_client._models import Run -from crawlee.events._types import Event, EventPersistStateData +from crawlee.events._types import Event, EventAbortingData, EventPersistStateData from ..._utils import poll_until_condition from apify import Actor @@ -112,6 +112,28 @@ async def test_fail_properly_deinitializes_actor(actor: _ActorType) -> None: assert actor._active is False +async def test_exit_from_event_listener_completes_cleanup() -> None: + """`Actor.exit()` called from an event listener runs cleanup instead of deadlocking into a RecursionError.""" + actor = Actor(exit_process=False) + await actor.init() + + exit_returned = False + + async def on_aborting(_data: EventAbortingData) -> None: + nonlocal exit_returned + await actor.exit(event_listeners_timeout=timedelta(seconds=1)) + exit_returned = True + + actor.on(Event.ABORTING, on_aborting) + actor.event_manager.emit(event=Event.ABORTING, event_data=EventAbortingData()) + + await poll_until_condition(lambda: not actor._active, timeout=5, poll_interval=0.1) + + assert exit_returned, 'Actor.exit() never returned inside the listener (deadlocked).' + assert actor._active is False + assert actor.event_manager.active is False + + async def test_failed_charging_manager_init_does_not_leak_event_manager() -> None: """Test that a failure in the charging manager's `__aenter__` also exits the already-entered event manager.""" actor = Actor()