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

Author: vdusek

src/apify/_actor.py

@@ -283,11 +283,24 @@ async def finalize() -> None:
             except Exception:
                 self.log.exception('Failed to save Actor state')
 
+        # When `exit()` / `fail()` is called from within an event listener (e.g. an `ABORTING` handler), that
+        # listener's own task is tracked in the event manager's listener-task set. The cleanup below waits for
+        # all listener tasks to finish -- directly, and again inside the event manager shutdown -- which would
+        # deadlock on the caller's own task and then, on the timeout cancellation, recurse into cancelling it,
+        # raising `RecursionError`. Detach it for the duration so the waits ignore it, then restore it so its
+        # wrapper can deregister it normally.
+        current_task = asyncio.current_task()
+        current_task_is_listener = current_task is not None and current_task in self.event_manager._listener_tasks  # noqa: SLF001
+        if current_task_is_listener:
+            self.event_manager._listener_tasks.discard(current_task)  # noqa: SLF001
+
         try:
             await asyncio.wait_for(finalize(), self._cleanup_timeout.total_seconds())
         except TimeoutError:
             self.log.exception('Actor cleanup timed out')
         finally:
+            if current_task_is_listener:
+                self.event_manager._listener_tasks.add(current_task)  # noqa: SLF001
             self._active = False
 
         if reraise_control_flow:

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,29 @@ async def test_fail_properly_deinitializes_actor(actor: _ActorType) -> None:
     assert actor._active is False
 
 
+async def test_exit_from_event_listener_completes_cleanup(caplog: pytest.LogCaptureFixture) -> 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
+    assert 'RecursionError' not in caplog.text
+
+
 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()