From 4ada3f46ed3d6f975f8b4ba496760b243e0bcaea Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Wed, 5 Aug 2026 12:45:03 +0200 Subject: [PATCH 1/4] fix: prevent deadlock when Actor.exit() is called from an event listener --- src/apify/_actor.py | 40 +++++++++++++++++++----- tests/unit/actor/test_actor_lifecycle.py | 26 +++++++++++++-- 2 files changed, 57 insertions(+), 9 deletions(-) diff --git a/src/apify/_actor.py b/src/apify/_actor.py index 9cf699e1..0ef47048 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 @@ -287,12 +288,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. @@ -1492,6 +1496,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 9f73df02..bf6f145c 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() From a290cd7719d489036f9f0bde74a7ee56f2d22bd2 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Wed, 5 Aug 2026 13:03:35 +0200 Subject: [PATCH 2/4] revert: drop SDK-side listener-detach workaround, require fixed crawlee lock crawlee's EventManager now handles this deadlock upstream (apify/crawlee-python#2088), so the SDK-side workaround is redundant. The lockfile is bumped to crawlee 1.9.1b4, which contains the fix; the regression test passes without any SDK-side code changes. The declared crawlee constraint in pyproject.toml stays >=1.8.0,<2.0.0 until crawlee ships a stable release with the fix. --- src/apify/_actor.py | 40 +++++++--------------------------------- uv.lock | 6 +++--- 2 files changed, 10 insertions(+), 36 deletions(-) diff --git a/src/apify/_actor.py b/src/apify/_actor.py index 0ef47048..9cf699e1 100644 --- a/src/apify/_actor.py +++ b/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.""" diff --git a/uv.lock b/uv.lock index 4b1767b5..2fad7763 100644 --- a/uv.lock +++ b/uv.lock @@ -585,7 +585,7 @@ toml = [ [[package]] name = "crawlee" -version = "1.8.3" +version = "1.9.1b4" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "async-timeout" }, @@ -602,9 +602,9 @@ dependencies = [ { name = "typing-extensions" }, { name = "yarl" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/10/b4/4c359c2e073c720b3960d2d41ab6e06b1ea0ff3d570e7c6467b899e787c1/crawlee-1.8.3.tar.gz", hash = "sha256:8f06e4bec07a5438126a22f5c6ae040f238e22eeaba1a4257de4240292419e6d", size = 316089, upload-time = "2026-07-20T07:07:51.617Z" } +sdist = { url = "https://files.pythonhosted.org/packages/5e/2c/8f6f5c7d44fbf7ed68d480e202144c309a896fbab9adb25bde5a9fea2c9d/crawlee-1.9.1b4.tar.gz", hash = "sha256:a50fa5355f6b3d03493977b6ad9aa0badbc7f0ab26dba7a281f3fef9a92b4cfd", size = 325031, upload-time = "2026-08-05T10:15:35.914Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/5a/85/05b8c4c73cad542e9aefe9a5568d03f73eb7ccb5f84524c47d066ce3bc85/crawlee-1.8.3-py3-none-any.whl", hash = "sha256:42c3e2404922a1ba51659e97ab31f0b451eeeed02faaf673d5ad4932f12bda7e", size = 403479, upload-time = "2026-07-20T07:07:49.96Z" }, + { url = "https://files.pythonhosted.org/packages/a5/f5/e6885878f73c7fbdb24681d8e1db8edc2949f7667f0c29e7be8a6fa8e76f/crawlee-1.9.1b4-py3-none-any.whl", hash = "sha256:c6bd7549d6027ad47074d913cb5b346a0b473f8e9f0efd57cd2e5ec15790e823", size = 411951, upload-time = "2026-08-05T10:15:34.212Z" }, ] [package.optional-dependencies] From d846455a0c2697748e33b416d01ef7c67cea307c Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Wed, 5 Aug 2026 13:23:56 +0200 Subject: [PATCH 3/4] fix: restore SDK-side listener-detach workaround for Python 3.11 --- src/apify/_actor.py | 40 +++++++++++++++++++++++++++++++++------- 1 file changed, 33 insertions(+), 7 deletions(-) diff --git a/src/apify/_actor.py b/src/apify/_actor.py index 9cf699e1..0ef47048 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 @@ -287,12 +288,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. @@ -1492,6 +1496,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.""" From d98c224352f68dedf876c4449ff4096364db63d3 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Wed, 5 Aug 2026 13:51:44 +0200 Subject: [PATCH 4/4] revert: drop SDK-side workaround, skip regression test on Python 3.11 --- src/apify/_actor.py | 40 +++++------------------- tests/unit/actor/test_actor_lifecycle.py | 8 +++++ 2 files changed, 15 insertions(+), 33 deletions(-) diff --git a/src/apify/_actor.py b/src/apify/_actor.py index 0ef47048..9cf699e1 100644 --- a/src/apify/_actor.py +++ b/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.""" diff --git a/tests/unit/actor/test_actor_lifecycle.py b/tests/unit/actor/test_actor_lifecycle.py index bf6f145c..0d81900b 100644 --- a/tests/unit/actor/test_actor_lifecycle.py +++ b/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)