Review comments

TODO: Figure out reason for stats difference in request_total_finished_duration

Author: Pijukatel

src/crawlee/_utils/recurring_task.py

@@ -7,6 +7,9 @@
 if TYPE_CHECKING:
     from collections.abc import Callable
     from datetime import timedelta
+    from types import TracebackType
+
+    from typing_extensions import Self
 
 logger = getLogger(__name__)
 
@@ -26,6 +29,18 @@ def __init__(self, func: Callable, delay: timedelta) -> None:
         self.delay = delay
         self.task: asyncio.Task | None = None
 
+    async def __aenter__(self) -> Self:
+        self.start()
+        return self
+
+    async def __aexit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_value: BaseException | None,
+        exc_traceback: TracebackType | None,
+    ) -> None:
+        await self.stop()
+
     async def _wrapper(self) -> None:
         """Continuously execute the provided function with the specified delay.
 

src/crawlee/crawlers/_basic/_basic_crawler.py

@@ -11,7 +11,7 @@
 from asyncio import CancelledError
 from collections.abc import AsyncGenerator, Awaitable, Callable, Iterable, Sequence
 from contextlib import AsyncExitStack, suppress
-from datetime import datetime, timedelta, timezone
+from datetime import timedelta
 from functools import partial
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Generic, Literal, cast
@@ -740,20 +740,11 @@ async def _run_crawler(self) -> None:
             for context in contexts_to_enter:
                 await exit_stack.enter_async_context(context)  # type: ignore[arg-type]
 
-            self._crawler_state_rec_task.start()
-            try:
+            async with self._crawler_state_rec_task:
                 await self._autoscaled_pool.run()
-            finally:
-                await self._crawler_state_rec_task.stop()
 
             # Emit PERSIST_STATE event when crawler is finishing to allow listeners to persist their state if needed
-            if not self.statistics.state.crawler_last_started_at:
-                raise RuntimeError('Statistics.state.crawler_last_started_at not set.')
-            run_duration = datetime.now(timezone.utc) - self.statistics.state.crawler_last_started_at
-            self._statistics.state.crawler_runtime = self.statistics.state.crawler_runtime + run_duration
-            self._service_locator.get_event_manager().emit(
-                event=Event.PERSIST_STATE, event_data=EventPersistStateData(is_migrating=False)
-            )
+            event_manager.emit(event=Event.PERSIST_STATE, event_data=EventPersistStateData(is_migrating=False))
 
     async def add_requests(
         self,

src/crawlee/statistics/_statistics.py

@@ -96,7 +96,7 @@ def __init__(
 
         self._state = RecoverableState(
             default_state=state_model(stats_id=self._id),
-            persist_state_key=persist_state_key or f'SDK_CRAWLER_STATISTICS_{self._id}',
+            persist_state_key=persist_state_key or f'__CRAWLER_STATISTICS_{self._id}',
             persistence_enabled=persistence_enabled,
             persist_state_kvs_name=persist_state_kvs_name,
             persist_state_kvs_factory=persist_state_kvs_factory,
@@ -187,7 +187,10 @@ async def __aexit__(
         if not self._active:
             raise RuntimeError(f'The {self.__class__.__name__} is not active.')
 
-        self._state.current_value.crawler_finished_at = datetime.now(timezone.utc)
+        if not self.state.crawler_last_started_at:
+            raise RuntimeError('Statistics.state.crawler_last_started_at not set.')
+        self.state.crawler_finished_at = datetime.now(timezone.utc)
+        self.state.crawler_runtime += self.state.crawler_finished_at - self.state.crawler_last_started_at
 
         await self._state.teardown()