fix: prevent deadlock when Actor.exit() is called from an event listener

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."""

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()