From fef2e2d85571f5fbe69488bf352f2eb4db34a223 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Tue, 21 Jul 2026 14:50:26 +0200 Subject: [PATCH 1/4] fix(events): prevent deadlock when closing or waiting for listeners from within a listener --- src/crawlee/events/_event_manager.py | 11 +++++- tests/unit/events/test_event_manager.py | 50 +++++++++++++++++++++++++ 2 files changed, 59 insertions(+), 2 deletions(-) diff --git a/src/crawlee/events/_event_manager.py b/src/crawlee/events/_event_manager.py index d44dd0f8c4..d0881e261c 100644 --- a/src/crawlee/events/_event_manager.py +++ b/src/crawlee/events/_event_manager.py @@ -202,7 +202,9 @@ async def listener_wrapper(event_data: EventData) -> None: ) finally: logger.debug('EventManager.on.listener_wrapper(): Removing listener task from the set...') - self._listener_tasks.remove(listener_task) + # Use `discard` rather than `remove`: the task may already be gone from the set if `__aexit__` + # cleared it while this listener was mid-flight (e.g. `Actor.exit()` called from the listener). + self._listener_tasks.discard(listener_task) self._listeners_to_wrappers[event][listener].append(listener_wrapper) self._event_emitter.add_listener(event.value, listener_wrapper) @@ -256,10 +258,15 @@ async def wait_for_all_listeners_to_complete(self, *, timeout: timedelta | None timeout: The maximum time to wait for the event listeners to finish. If they do not complete within the specified timeout, they will be canceled. """ + # Exclude the calling task from the wait set. When this method is reached from within an event listener + # (e.g. `Actor.exit()` invoked inside an `ABORTING` handler), that listener runs in a task which is itself + # present in `_listener_tasks`. Awaiting it here would make the task wait for its own completion and deadlock. + current_task = asyncio.current_task() async def wait_for_listeners() -> None: """Gathers all listener tasks and awaits their completion, logging any exceptions encountered.""" - results = await asyncio.gather(*self._listener_tasks, return_exceptions=True) + listener_tasks = [task for task in self._listener_tasks if task is not current_task] + results = await asyncio.gather(*listener_tasks, return_exceptions=True) for result in results: if isinstance(result, Exception): logger.exception('Event listener raised an exception.', exc_info=result) diff --git a/tests/unit/events/test_event_manager.py b/tests/unit/events/test_event_manager.py index 2b7b7cd10c..80bfef657a 100644 --- a/tests/unit/events/test_event_manager.py +++ b/tests/unit/events/test_event_manager.py @@ -206,6 +206,56 @@ async def test_methods_raise_error_when_not_active(event_system_info_data: Event assert event_manager.active is True +async def test_wait_for_all_listeners_from_within_a_listener_does_not_deadlock( + event_manager: EventManager, + event_system_info_data: EventSystemInfoData, +) -> None: + """Waiting for all listeners from within a listener must not make the listener wait for its own task.""" + completed = asyncio.Event() + + async def waiting_listener(_: Any) -> None: + await event_manager.wait_for_all_listeners_to_complete() + completed.set() + + event_manager.on(event=Event.SYSTEM_INFO, listener=waiting_listener) + event_manager.emit(event=Event.SYSTEM_INFO, event_data=event_system_info_data) + + await asyncio.wait_for(completed.wait(), timeout=5) + + +async def test_close_from_within_a_listener_does_not_deadlock_or_error( + event_system_info_data: EventSystemInfoData, +) -> None: + """Closing the event manager from within a listener (as `Actor.exit()` does) must not deadlock or raise.""" + event_manager = EventManager() + await event_manager.__aenter__() + + # Capture exceptions raised inside listener wrappers - pyee re-emits them on the `error` event. + errors: list[BaseException] = [] + event_manager._event_emitter.add_listener('error', errors.append) + + closed = asyncio.Event() + + async def closing_listener(_: Any) -> None: + await event_manager.__aexit__(None, None, None) + closed.set() + + event_manager.on(event=Event.SYSTEM_INFO, listener=closing_listener) + event_manager.emit(event=Event.SYSTEM_INFO, event_data=event_system_info_data) + + try: + await asyncio.wait_for(closed.wait(), timeout=5) + # Let the listener wrapper run its `finally` block that discards the completed task. + await asyncio.sleep(0.1) + finally: + if event_manager.active: + await event_manager.__aexit__(None, None, None) + + assert errors == [] + assert event_manager.active is False + assert len(event_manager._listener_tasks) == 0 + + async def test_event_manager_in_context_persistence() -> None: """Test that entering the `EventManager` context emits persist state event at least once.""" event_manager = EventManager() From 81b59d5de0e2ba6da697d91c6fe63ba4130d5276 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Tue, 21 Jul 2026 14:55:46 +0200 Subject: [PATCH 2/4] style(events): make deadlock-fix comments more concise --- src/crawlee/events/_event_manager.py | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/src/crawlee/events/_event_manager.py b/src/crawlee/events/_event_manager.py index d0881e261c..bcf8104ca6 100644 --- a/src/crawlee/events/_event_manager.py +++ b/src/crawlee/events/_event_manager.py @@ -202,8 +202,7 @@ async def listener_wrapper(event_data: EventData) -> None: ) finally: logger.debug('EventManager.on.listener_wrapper(): Removing listener task from the set...') - # Use `discard` rather than `remove`: the task may already be gone from the set if `__aexit__` - # cleared it while this listener was mid-flight (e.g. `Actor.exit()` called from the listener). + # `discard` not `remove`: `__aexit__` may have already cleared the set while this listener ran. self._listener_tasks.discard(listener_task) self._listeners_to_wrappers[event][listener].append(listener_wrapper) @@ -258,9 +257,7 @@ async def wait_for_all_listeners_to_complete(self, *, timeout: timedelta | None timeout: The maximum time to wait for the event listeners to finish. If they do not complete within the specified timeout, they will be canceled. """ - # Exclude the calling task from the wait set. When this method is reached from within an event listener - # (e.g. `Actor.exit()` invoked inside an `ABORTING` handler), that listener runs in a task which is itself - # present in `_listener_tasks`. Awaiting it here would make the task wait for its own completion and deadlock. + # Exclude the calling task: when called from within a listener, awaiting it here would deadlock (self-await). current_task = asyncio.current_task() async def wait_for_listeners() -> None: From 68c74da27439831ecd58d3d8ced66611245432a2 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Tue, 21 Jul 2026 15:22:21 +0200 Subject: [PATCH 3/4] fix(events): also prevent deadlock between multiple listeners waiting at once --- src/crawlee/events/_event_manager.py | 20 ++++-- tests/unit/events/test_event_manager.py | 87 +++++++++++++++++++++---- 2 files changed, 91 insertions(+), 16 deletions(-) diff --git a/src/crawlee/events/_event_manager.py b/src/crawlee/events/_event_manager.py index bcf8104ca6..5255b7bf4e 100644 --- a/src/crawlee/events/_event_manager.py +++ b/src/crawlee/events/_event_manager.py @@ -81,6 +81,10 @@ def __init__( # Listeners are wrapped inside asyncio.Task. Store their references here so that we can wait for them to finish. self._listener_tasks: set[asyncio.Task] = set() + # Tasks currently blocked inside `wait_for_all_listeners_to_complete`. They are excluded when gathering + # listener tasks, otherwise a waiter would await itself or another waiter and deadlock. + self._waiting_listener_tasks: set[asyncio.Task] = set() + # Store the mapping between events, listeners and their wrappers in the following way: # event -> listener -> [wrapped_listener_1, wrapped_listener_2, ...] self._listeners_to_wrappers: dict[Event, dict[EventListener[Any], list[WrappedListener]]] = defaultdict( @@ -257,12 +261,16 @@ async def wait_for_all_listeners_to_complete(self, *, timeout: timedelta | None timeout: The maximum time to wait for the event listeners to finish. If they do not complete within the specified timeout, they will be canceled. """ - # Exclude the calling task: when called from within a listener, awaiting it here would deadlock (self-await). - current_task = asyncio.current_task() + # Register this call's task as a waiter. A waiting task cannot finish until the listeners it awaits do, + # so no waiter may await another waiter (nor itself) - otherwise they would deadlock. This happens when + # a listener waits for or closes the event manager from within itself (as `Actor.exit()` does). + waiting_task = asyncio.current_task() + if waiting_task is not None: + self._waiting_listener_tasks.add(waiting_task) async def wait_for_listeners() -> None: """Gathers all listener tasks and awaits their completion, logging any exceptions encountered.""" - listener_tasks = [task for task in self._listener_tasks if task is not current_task] + listener_tasks = [task for task in self._listener_tasks if task not in self._waiting_listener_tasks] results = await asyncio.gather(*listener_tasks, return_exceptions=True) for result in results: if isinstance(result, Exception): @@ -270,7 +278,11 @@ async def wait_for_listeners() -> None: tasks = [asyncio.create_task(wait_for_listeners(), name=f'Task-{wait_for_listeners.__name__}')] - await wait_for_all_tasks_for_finish(tasks=tasks, logger=logger, timeout=timeout) + try: + await wait_for_all_tasks_for_finish(tasks=tasks, logger=logger, timeout=timeout) + finally: + if waiting_task is not None: + self._waiting_listener_tasks.discard(waiting_task) async def _emit_persist_state_event(self) -> None: """Emit a persist state event with the given migration status.""" diff --git a/tests/unit/events/test_event_manager.py b/tests/unit/events/test_event_manager.py index 80bfef657a..3491554211 100644 --- a/tests/unit/events/test_event_manager.py +++ b/tests/unit/events/test_event_manager.py @@ -2,6 +2,7 @@ import asyncio import logging +from contextlib import suppress from datetime import timedelta from functools import update_wrapper from typing import TYPE_CHECKING, Any @@ -210,17 +211,56 @@ async def test_wait_for_all_listeners_from_within_a_listener_does_not_deadlock( event_manager: EventManager, event_system_info_data: EventSystemInfoData, ) -> None: - """Waiting for all listeners from within a listener must not make the listener wait for its own task.""" - completed = asyncio.Event() + """Waiting from within a listener must not self-await, yet must still await the other listeners.""" + other_listener_done = asyncio.Event() + waiter_done = asyncio.Event() + other_done_when_wait_returned: bool | None = None + + async def other_listener(_: Any) -> None: + await asyncio.sleep(0.2) + other_listener_done.set() async def waiting_listener(_: Any) -> None: + nonlocal other_done_when_wait_returned await event_manager.wait_for_all_listeners_to_complete() - completed.set() + other_done_when_wait_returned = other_listener_done.is_set() + waiter_done.set() + event_manager.on(event=Event.SYSTEM_INFO, listener=other_listener) event_manager.on(event=Event.SYSTEM_INFO, listener=waiting_listener) event_manager.emit(event=Event.SYSTEM_INFO, event_data=event_system_info_data) - await asyncio.wait_for(completed.wait(), timeout=5) + await asyncio.wait_for(waiter_done.wait(), timeout=5) + + # No self-await deadlock, and the wait must have blocked until the co-registered listener finished. + assert other_done_when_wait_returned is True + assert other_listener_done.is_set() + + +async def test_wait_from_within_multiple_listeners_does_not_deadlock( + event_manager: EventManager, + event_system_info_data: EventSystemInfoData, +) -> None: + """Several listeners each waiting for all listeners at once must not deadlock one another.""" + first_done = asyncio.Event() + second_done = asyncio.Event() + + async def first_waiting_listener(_: Any) -> None: + await event_manager.wait_for_all_listeners_to_complete() + first_done.set() + + async def second_waiting_listener(_: Any) -> None: + await event_manager.wait_for_all_listeners_to_complete() + second_done.set() + + event_manager.on(event=Event.SYSTEM_INFO, listener=first_waiting_listener) + event_manager.on(event=Event.SYSTEM_INFO, listener=second_waiting_listener) + event_manager.emit(event=Event.SYSTEM_INFO, event_data=event_system_info_data) + + await asyncio.wait_for(asyncio.gather(first_done.wait(), second_done.wait()), timeout=5) + + assert first_done.is_set() + assert second_done.is_set() async def test_close_from_within_a_listener_does_not_deadlock_or_error( @@ -230,28 +270,51 @@ async def test_close_from_within_a_listener_does_not_deadlock_or_error( event_manager = EventManager() await event_manager.__aenter__() - # Capture exceptions raised inside listener wrappers - pyee re-emits them on the `error` event. - errors: list[BaseException] = [] - event_manager._event_emitter.add_listener('error', errors.append) + # A wrapper that finalizes after close raises through pyee onto the event loop (its `error` listener is + # gone by then, removed during close), so watch both channels to catch a stray exception on finalize. + emitter_errors: list[BaseException] = [] + event_manager._event_emitter.add_listener('error', emitter_errors.append) + loop_errors: list[dict[str, Any]] = [] + asyncio.get_running_loop().set_exception_handler(lambda _loop, context: loop_errors.append(context)) closed = asyncio.Event() + other_listener_done = asyncio.Event() + + async def other_listener(_: Any) -> None: + await asyncio.sleep(0.2) + other_listener_done.set() async def closing_listener(_: Any) -> None: await event_manager.__aexit__(None, None, None) closed.set() + # Register a regular listener too, so closing must await a concurrently-running listener (the real + # `Actor.exit()` shape) and exercise the task-set cleanup while another listener is still in flight. + event_manager.on(event=Event.SYSTEM_INFO, listener=other_listener) event_manager.on(event=Event.SYSTEM_INFO, listener=closing_listener) + + tasks_before = asyncio.all_tasks() event_manager.emit(event=Event.SYSTEM_INFO, event_data=event_system_info_data) try: await asyncio.wait_for(closed.wait(), timeout=5) - # Let the listener wrapper run its `finally` block that discards the completed task. - await asyncio.sleep(0.1) + # Deterministically drain the listener-wrapper tasks so their `finally` blocks (and any exception + # they raise on finalize) run before we assert - no arbitrary sleep. + spawned = asyncio.all_tasks() - tasks_before - {asyncio.current_task()} + if spawned: + await asyncio.wait(spawned) finally: + # On a regression the listener may deadlock; cap the cleanup so the primary failure surfaces instead + # of hanging indefinitely. if event_manager.active: - await event_manager.__aexit__(None, None, None) - - assert errors == [] + with suppress(Exception): + await asyncio.wait_for(event_manager.__aexit__(None, None, None), timeout=5) + + # The discard-not-remove fix means no wrapper raises on finalize; the `remove` regression would surface + # on one of these channels (whichever fires depends on whether the `error` listener is still registered). + assert emitter_errors == [] + assert loop_errors == [] + assert other_listener_done.is_set() assert event_manager.active is False assert len(event_manager._listener_tasks) == 0 From c5405e213f71b84fe5435f12a117d08dd4986d77 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Tue, 21 Jul 2026 15:25:42 +0200 Subject: [PATCH 4/4] style(events): make comments more concise --- src/crawlee/events/_event_manager.py | 10 ++++------ tests/unit/events/test_event_manager.py | 16 ++++++---------- 2 files changed, 10 insertions(+), 16 deletions(-) diff --git a/src/crawlee/events/_event_manager.py b/src/crawlee/events/_event_manager.py index 5255b7bf4e..2132426198 100644 --- a/src/crawlee/events/_event_manager.py +++ b/src/crawlee/events/_event_manager.py @@ -81,8 +81,7 @@ def __init__( # Listeners are wrapped inside asyncio.Task. Store their references here so that we can wait for them to finish. self._listener_tasks: set[asyncio.Task] = set() - # Tasks currently blocked inside `wait_for_all_listeners_to_complete`. They are excluded when gathering - # listener tasks, otherwise a waiter would await itself or another waiter and deadlock. + # Tasks currently blocked in `wait_for_all_listeners_to_complete`; excluded when gathering to avoid deadlock. self._waiting_listener_tasks: set[asyncio.Task] = set() # Store the mapping between events, listeners and their wrappers in the following way: @@ -206,7 +205,7 @@ async def listener_wrapper(event_data: EventData) -> None: ) finally: logger.debug('EventManager.on.listener_wrapper(): Removing listener task from the set...') - # `discard` not `remove`: `__aexit__` may have already cleared the set while this listener ran. + # `discard`, not `remove`: `__aexit__` may have cleared the set while this listener ran. self._listener_tasks.discard(listener_task) self._listeners_to_wrappers[event][listener].append(listener_wrapper) @@ -261,9 +260,8 @@ async def wait_for_all_listeners_to_complete(self, *, timeout: timedelta | None timeout: The maximum time to wait for the event listeners to finish. If they do not complete within the specified timeout, they will be canceled. """ - # Register this call's task as a waiter. A waiting task cannot finish until the listeners it awaits do, - # so no waiter may await another waiter (nor itself) - otherwise they would deadlock. This happens when - # a listener waits for or closes the event manager from within itself (as `Actor.exit()` does). + # A waiter can't finish until the listeners it awaits do, so waiters must never await each other or + # themselves - this is what happens when a listener waits or closes from within itself. waiting_task = asyncio.current_task() if waiting_task is not None: self._waiting_listener_tasks.add(waiting_task) diff --git a/tests/unit/events/test_event_manager.py b/tests/unit/events/test_event_manager.py index 3491554211..584bbe9ced 100644 --- a/tests/unit/events/test_event_manager.py +++ b/tests/unit/events/test_event_manager.py @@ -270,8 +270,8 @@ async def test_close_from_within_a_listener_does_not_deadlock_or_error( event_manager = EventManager() await event_manager.__aenter__() - # A wrapper that finalizes after close raises through pyee onto the event loop (its `error` listener is - # gone by then, removed during close), so watch both channels to catch a stray exception on finalize. + # A wrapper finalizing after close raises onto the loop (its `error` listener is gone by then), so watch + # both channels for a stray exception. emitter_errors: list[BaseException] = [] event_manager._event_emitter.add_listener('error', emitter_errors.append) loop_errors: list[dict[str, Any]] = [] @@ -288,8 +288,7 @@ async def closing_listener(_: Any) -> None: await event_manager.__aexit__(None, None, None) closed.set() - # Register a regular listener too, so closing must await a concurrently-running listener (the real - # `Actor.exit()` shape) and exercise the task-set cleanup while another listener is still in flight. + # A second listener makes close await a concurrently-running listener - the real `Actor.exit()` shape. event_manager.on(event=Event.SYSTEM_INFO, listener=other_listener) event_manager.on(event=Event.SYSTEM_INFO, listener=closing_listener) @@ -298,20 +297,17 @@ async def closing_listener(_: Any) -> None: try: await asyncio.wait_for(closed.wait(), timeout=5) - # Deterministically drain the listener-wrapper tasks so their `finally` blocks (and any exception - # they raise on finalize) run before we assert - no arbitrary sleep. + # Drain the wrapper tasks so their `finally` blocks run before we assert - no arbitrary sleep. spawned = asyncio.all_tasks() - tasks_before - {asyncio.current_task()} if spawned: await asyncio.wait(spawned) finally: - # On a regression the listener may deadlock; cap the cleanup so the primary failure surfaces instead - # of hanging indefinitely. + # Cap the cleanup so a regressed deadlock surfaces the real failure instead of hanging. if event_manager.active: with suppress(Exception): await asyncio.wait_for(event_manager.__aexit__(None, None, None), timeout=5) - # The discard-not-remove fix means no wrapper raises on finalize; the `remove` regression would surface - # on one of these channels (whichever fires depends on whether the `error` listener is still registered). + # With `discard` no wrapper raises on finalize; the `remove` regression would surface on one of these. assert emitter_errors == [] assert loop_errors == [] assert other_listener_done.is_set()