revert: drop SDK-side workaround, skip regression test on Python 3.11

Author: vdusek

src/apify/_actor.py

@@ -4,7 +4,6 @@
 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
@@ -42,7 +41,7 @@
 
 if TYPE_CHECKING:
     import logging
-    from collections.abc import Callable, Iterator, MutableMapping
+    from collections.abc import Callable, MutableMapping
     from decimal import Decimal
     from types import TracebackType
     from typing import Self
@@ -288,15 +287,12 @@ async def finalize() -> None:
             except Exception:
                 self.log.exception('Failed to save Actor state')
 
-        # `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
+        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.
@@ -1496,28 +1492,6 @@ 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,6 +4,7 @@
 import contextlib
 import json
 import logging
+import sys
 from datetime import UTC, datetime, timedelta
 from typing import TYPE_CHECKING, Any
 from unittest import mock
@@ -112,6 +113,13 @@ async def test_fail_properly_deinitializes_actor(actor: _ActorType) -> None:
     assert actor._active is False
 
 
+@pytest.mark.skipif(
+    sys.version_info < (3, 12),
+    reason=(
+        'On Python 3.11, `asyncio.wait_for` runs the awaited coroutine in a separate task, which defeats '
+        "crawlee's own self-wait detection in `EventManager.wait_for_all_listeners_to_complete` and deadlocks."
+    ),
+)
 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)