From f171a4eda6dfb96fac01265b1da778d2f17971aa Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Tue, 21 Jul 2026 11:43:49 +0200 Subject: [PATCH 1/3] fix: prevent deadlock when Actor.exit() is called from an event listener --- src/apify/_actor.py | 13 ++++++++++++ tests/unit/actor/test_actor_lifecycle.py | 27 ++++++++++++++++++++++-- 2 files changed, 38 insertions(+), 2 deletions(-) diff --git a/src/apify/_actor.py b/src/apify/_actor.py index c75cbf26..141f6b33 100644 --- a/src/apify/_actor.py +++ b/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: diff --git a/tests/unit/actor/test_actor_lifecycle.py b/tests/unit/actor/test_actor_lifecycle.py index a702cd09..e9e64438 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,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() From 1ac85775c50e47eead10574f45e81b66b72354bd Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Tue, 21 Jul 2026 12:51:53 +0200 Subject: [PATCH 2/3] refactor: extract listener-task detach in Actor cleanup into a context manager --- src/apify/_actor.py | 58 ++++++++++++++++-------- tests/unit/actor/test_actor_lifecycle.py | 3 +- 2 files changed, 39 insertions(+), 22 deletions(-) diff --git a/src/apify/_actor.py b/src/apify/_actor.py index 141f6b33..1f3d4f7c 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,25 +284,15 @@ 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 + # `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. @@ -1501,6 +1492,33 @@ 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. + + When `exit()` / `fail()` is called from within an event listener (e.g. an `ABORTING` handler), the current + task is that listener's own task, which the event manager tracks in its listener-task set. Actor cleanup + 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`. Detaching it for the duration makes those waits ignore it; restoring it + afterwards lets the listener wrapper deregister it normally. + + Only a direct call on the listener's own task is handled. A call from a task the listener spawns itself + (e.g. via `asyncio.create_task` or `asyncio.gather`) is not detected, because asyncio does not expose + task ancestry; robustly covering that would require reworking listener registration or upstream support + in the event manager. + """ + 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 e9e64438..0324e6fa 100644 --- a/tests/unit/actor/test_actor_lifecycle.py +++ b/tests/unit/actor/test_actor_lifecycle.py @@ -112,7 +112,7 @@ 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: +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() @@ -132,7 +132,6 @@ async def on_aborting(_data: EventAbortingData) -> None: 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: From b2a2255692013b74d5ed0ad1e4ebfbec1b06400b Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Tue, 21 Jul 2026 13:06:08 +0200 Subject: [PATCH 3/3] docs: shorten _detach_current_listener_task docstring --- src/apify/_actor.py | 17 ++++++----------- 1 file changed, 6 insertions(+), 11 deletions(-) diff --git a/src/apify/_actor.py b/src/apify/_actor.py index 1f3d4f7c..3db81bbc 100644 --- a/src/apify/_actor.py +++ b/src/apify/_actor.py @@ -1496,17 +1496,12 @@ def _get_remaining_time(self) -> timedelta | None: def _detach_current_listener_task(self) -> Iterator[None]: """Temporarily remove the current task from the event manager's listener-task set. - When `exit()` / `fail()` is called from within an event listener (e.g. an `ABORTING` handler), the current - task is that listener's own task, which the event manager tracks in its listener-task set. Actor cleanup - 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`. Detaching it for the duration makes those waits ignore it; restoring it - afterwards lets the listener wrapper deregister it normally. - - Only a direct call on the listener's own task is handled. A call from a task the listener spawns itself - (e.g. via `asyncio.create_task` or `asyncio.gather`) is not detected, because asyncio does not expose - task ancestry; robustly covering that would require reworking listener registration or upstream support - in the event manager. + 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()