fix(events): correct the listener invocation, removal, and context release

pyproject.toml

@@ -43,7 +43,6 @@ dependencies = [
     "psutil>=6.0.0",
     "pydantic-settings>=2.12.0",
     "pydantic>=2.11.0",
-    "pyee>=9.0.0",
     "tldextract>=5.3.0",
     "typing-extensions>=4.10.0",
     "yarl>=1.18.0",

src/crawlee/_utils/wait.py

@@ -46,7 +46,7 @@ async def wait_for(
     raise RuntimeError('Unreachable code')
 
 
-async def wait_for_all_tasks_for_finish(
+async def wait_for_all_tasks_to_finish(
     tasks: Sequence[asyncio.Task],
     *,
     logger: Logger,

src/crawlee/events/_local_event_manager.py

@@ -17,7 +17,7 @@
 if TYPE_CHECKING:
     from types import TracebackType
 
-    from typing_extensions import Unpack
+    from typing_extensions import Self, Unpack
 
 logger = getLogger(__name__)
 
@@ -45,15 +45,16 @@ def __init__(
             system_info_interval: Interval at which `SystemInfo` events are emitted.
             event_manager_options: Additional options for the parent class.
         """
+        super().__init__(**event_manager_options)
+
         self._system_info_interval = system_info_interval
+        """Interval between the emitted `SystemInfo` events."""
 
-        # Recurring task for emitting system info events.
         self._emit_system_info_event_rec_task = RecurringTask(
             func=self._emit_system_info_event,
             delay=self._system_info_interval,
         )
-
-        super().__init__(**event_manager_options)
+        """Recurring task emitting the `SystemInfo` events."""
 
     @classmethod
     def from_config(cls, config: Configuration | None = None) -> LocalEventManager:
@@ -69,7 +70,7 @@ def from_config(cls, config: Configuration | None = None) -> LocalEventManager:
             persist_state_interval=config.persist_state_interval,
         )
 
-    async def __aenter__(self) -> LocalEventManager:
+    async def __aenter__(self) -> Self:
         """Initialize the local event manager upon entering the async context.
 
         It starts emitting system info events at regular intervals.
@@ -98,8 +99,12 @@ async def __aexit__(
 
     async def _emit_system_info_event(self) -> None:
         """Emit a system info event with the current CPU and memory usage."""
-        cpu_info = await asyncio.to_thread(get_cpu_info)
-        memory_info = await asyncio.to_thread(get_memory_info)
+        # Both readings block the thread they run in - `get_cpu_info` even samples the CPU utilization over a short
+        # interval - so run them concurrently instead of one after the other.
+        cpu_info, memory_info = await asyncio.gather(
+            asyncio.to_thread(get_cpu_info),
+            asyncio.to_thread(get_memory_info),
+        )
 
         event_data = EventSystemInfoData(cpu_info=cpu_info, memory_info=memory_info)
         self.emit(event=Event.SYSTEM_INFO, event_data=event_data)

src/crawlee/storages/_request_queue.py

@@ -9,7 +9,7 @@
 
 from crawlee import Request, service_locator
 from crawlee._utils.docs import docs_group
-from crawlee._utils.wait import wait_for_all_tasks_for_finish
+from crawlee._utils.wait import wait_for_all_tasks_to_finish
 from crawlee.request_loaders import RequestManager
 from crawlee.storage_clients.models import AddRequestsResponse
 
@@ -242,7 +242,7 @@ async def _process_remaining_batches() -> None:
 
         # Wait for all tasks to finish if requested
         if wait_for_all_requests_to_be_added:
-            await wait_for_all_tasks_for_finish(
+            await wait_for_all_tasks_to_finish(
                 (remaining_batches_task,),
                 logger=logger,
                 timeout=wait_for_all_requests_to_be_added_timeout,