diff --git a/docs/upgrading/upgrading_to_v2.md b/docs/upgrading/upgrading_to_v2.md index 894d2a8567..090bfc7c95 100644 --- a/docs/upgrading/upgrading_to_v2.md +++ b/docs/upgrading/upgrading_to_v2.md @@ -15,7 +15,11 @@ We drop support for Python 3.10, which reached its end of life in October 2026. ## `StatisticsState.crawler_runtime` is read-only -`StatisticsState.crawler_runtime` is computed from the crawler's start and finish timestamps. In v1, assigning to it only emitted a `DeprecationWarning` and the value was ignored. In v2, the property has no setter, so an assignment raises `AttributeError`. Remove any such assignments. +`StatisticsState.crawler_runtime` is computed from the runtime accumulated by previous runs and the start timestamp of the run in progress. In v1, assigning to it only emitted a `DeprecationWarning` and the value was ignored. In v2, the property has no setter, so an assignment raises `AttributeError`. Remove any such assignments. + +## `StatisticsState.crawler_runtime_for_serialization` is removed + +`StatisticsState.crawler_runtime_for_serialization` is removed, and the persisted `crawlerRuntimeMillis` value is a number of milliseconds instead of an ISO 8601 duration string. Read the runtime from the `crawler_runtime` property instead. Statistics persisted by v1 are still restored. ## `RequestQueueClient.is_finished` is abstract diff --git a/src/crawlee/statistics/_models.py b/src/crawlee/statistics/_models.py index 1db25cd5ab..8a7d2a3873 100644 --- a/src/crawlee/statistics/_models.py +++ b/src/crawlee/statistics/_models.py @@ -5,7 +5,7 @@ from datetime import UTC, datetime, timedelta from typing import TYPE_CHECKING, Annotated, Any -from pydantic import BaseModel, ConfigDict, Field, PlainSerializer, PlainValidator, computed_field +from pydantic import BaseModel, ConfigDict, Field, PlainSerializer, PlainValidator, computed_field, field_serializer from pydantic.alias_generators import to_camel from typing_extensions import override @@ -101,25 +101,25 @@ class StatisticsState(BaseModel): ), ] = {} - # Used to track the crawler runtime, that had already been persisted. This is the runtime from previous runs. - _runtime_offset: Annotated[timedelta, Field(exclude=True)] = timedelta() - - def model_post_init(self, /, __context: Any) -> None: - self._runtime_offset = self.crawler_runtime or self._runtime_offset + accumulated_runtime: Annotated[timedelta_ms, Field(alias='crawlerRuntimeMillis')] = timedelta() + """The runtime accumulated by the previous runs of the crawler, excluding the run in progress.""" @property def crawler_runtime(self) -> timedelta: - if self.crawler_last_started_at: - finished_at = self.crawler_finished_at or datetime.now(UTC) - return self._runtime_offset + finished_at - self.crawler_last_started_at - return self._runtime_offset - - @computed_field(alias='crawlerRuntimeMillis') - def crawler_runtime_for_serialization(self) -> timedelta: - if self.crawler_last_started_at: - finished_at = self.crawler_finished_at or datetime.now(UTC) - return self._runtime_offset + finished_at - self.crawler_last_started_at - return self._runtime_offset + """The total runtime of the crawler, accumulated over all its runs.""" + if self.crawler_last_started_at is None: + return self.accumulated_runtime + if self.stats_persisted_at is not None and self.crawler_last_started_at <= self.stats_persisted_at: + # A restored snapshot already checkpoints this run in `accumulated_runtime`, do not count it twice. + return self.accumulated_runtime + + finished_at = self.crawler_finished_at or datetime.now(UTC) + return self.accumulated_runtime + max(timedelta(), finished_at - self.crawler_last_started_at) + + @field_serializer('accumulated_runtime') + def _serialize_accumulated_runtime(self, _value: timedelta) -> int: + """Serialize the total runtime including the run in progress, so that every persist checkpoints it.""" + return round(self.crawler_runtime.total_seconds() * 1000) @computed_field(alias='requestTotalDurationMillis', return_type=timedelta_ms) @property diff --git a/src/crawlee/statistics/_statistics.py b/src/crawlee/statistics/_statistics.py index 0e39aa3557..02c1e4e90a 100644 --- a/src/crawlee/statistics/_statistics.py +++ b/src/crawlee/statistics/_statistics.py @@ -173,10 +173,15 @@ async def __aenter__(self) -> Self: raise RuntimeError(f'The {self.__class__.__name__} is already active.') await self._state.initialize() - # Reset `crawler_finished_at` to indicate a new run in progress. + + # The restored timestamps belong to the previous run. Clear them, so that the downtime since then (e.g. a + # migration or resurrection) doesn't count towards the runtime, and a clock that is behind the previous + # host's doesn't make `crawler_runtime` treat the new run as already checkpointed. + self.state.crawler_last_started_at = None self.state.crawler_finished_at = None + self.state.stats_persisted_at = None - # Start periodic logging and let it print initial state before activation. + # Start periodic logging and let it print the restored state before the new run starts being measured. self._periodic_logger.start() await asyncio.sleep(0.01) self._active = True @@ -199,10 +204,7 @@ async def __aexit__( if not self._active: raise RuntimeError(f'The {self.__class__.__name__} is not active.') - if not self.state.crawler_last_started_at: - raise RuntimeError('Statistics.state.crawler_last_started_at not set.') - - # Stop logging and deactivate the statistics to prevent further changes to crawler_runtime + # Stop logging; the finish timestamp freezes the measured run before the final persist. await self._periodic_logger.stop() self.state.crawler_finished_at = datetime.now(UTC) self._active = False diff --git a/tests/unit/_statistics/test_persistence.py b/tests/unit/_statistics/test_persistence.py index 3ee559e40c..696a7a25f0 100644 --- a/tests/unit/_statistics/test_persistence.py +++ b/tests/unit/_statistics/test_persistence.py @@ -1,5 +1,9 @@ from __future__ import annotations +import asyncio +import logging +from datetime import UTC, datetime, timedelta + import pytest from crawlee.statistics import Statistics, StatisticsState @@ -18,6 +22,206 @@ async def test_basic_persistence() -> None: assert statistics.state.requests_failed == 42 +async def test_first_periodic_log_of_fresh_run_reports_zero_runtime(caplog: pytest.LogCaptureFixture) -> None: + """The first periodic log of a fresh run reports a runtime of exactly zero.""" + caplog.set_level(logging.INFO) + log_message = 'Fresh statistics' + + async with Statistics.with_default_state(log_message=log_message, statistics_log_format='inline'): + pass + + periodic_records = [record for record in caplog.records if record.message == log_message] + assert periodic_records + assert periodic_records[0].crawler_runtime == 0 # ty: ignore[unresolved-attribute] + + +async def test_periodic_log_after_resume_excludes_downtime(caplog: pytest.LogCaptureFixture) -> None: + """The first periodic log of a resumed run reports exactly the previous runtime, without the downtime.""" + caplog.set_level(logging.INFO) + key = 'statistics_downtime_clean' + log_message = 'Statistics after resume' + downtime = timedelta(hours=2) + + async with Statistics.with_default_state(persistence_enabled=True, persist_state_key=key): + pass + + # Simulate a resurrection after two hours of downtime by shifting the persisted timestamps into the past. + kvs = await KeyValueStore.open() + stored_state = await kvs.get_value(key) + for field in ('crawlerStartedAt', 'crawlerLastStartTimestamp', 'crawlerFinishedAt', 'statsPersistedAt'): + stored_state[field] = (datetime.fromisoformat(stored_state[field]) - downtime).isoformat() + await kvs.set_value(key, stored_state) + + caplog.clear() + async with Statistics.with_default_state( + persistence_enabled=True, + persist_state_key=key, + log_message=log_message, + statistics_log_format='inline', + ): + pass + + periodic_records = [record for record in caplog.records if record.message == log_message] + assert periodic_records + first_logged_runtime = timedelta(seconds=periodic_records[0].crawler_runtime) # ty: ignore[unresolved-attribute] + previous_runtime = timedelta(milliseconds=stored_state['crawlerRuntimeMillis']) + assert abs(first_logged_runtime - previous_runtime) < timedelta(milliseconds=1) + + +async def test_runtime_accumulates_over_multiple_resurrections() -> None: + """The persisted total runtime is restored on resume, so runs before the last one still count.""" + key = 'statistics_accumulated_runtime' + now = datetime.now(UTC) + last_run_start = now - timedelta(hours=1) + last_run_duration = timedelta(seconds=5) + total_runtime = timedelta(seconds=30) + + # State persisted after a clean run whose own segment took 5s, with 30s of runtime accumulated in total. + kvs = await KeyValueStore.open() + await kvs.set_value( + key, + { + 'crawlerStartedAt': (now - timedelta(hours=2)).isoformat(), + 'crawlerLastStartTimestamp': last_run_start.isoformat(), + 'crawlerFinishedAt': (last_run_start + last_run_duration).isoformat(), + 'statsPersistedAt': (last_run_start + last_run_duration).isoformat(), + 'crawlerRuntimeMillis': total_runtime.total_seconds() * 1000, + }, + ) + + async with Statistics.with_default_state(persistence_enabled=True, persist_state_key=key) as statistics: + runtime = statistics.state.crawler_runtime + + assert total_runtime <= runtime < total_runtime + timedelta(minutes=1) + + +async def test_runtime_after_unclean_shutdown_excludes_downtime() -> None: + """A state checkpointed mid-run restores the runtime up to the last persist, without the downtime that follows.""" + key = 'statistics_downtime_unclean' + now = datetime.now(UTC) + downtime = timedelta(hours=2) + previous_runtime = timedelta(seconds=10) + + # State persisted by a periodic persist during a run that never finished cleanly (migration or abort): + # there is no finish timestamp, only the runtime checkpointed at the last persist. + kvs = await KeyValueStore.open() + await kvs.set_value( + key, + { + 'crawlerStartedAt': (now - downtime - previous_runtime).isoformat(), + 'crawlerLastStartTimestamp': (now - downtime - previous_runtime).isoformat(), + 'crawlerFinishedAt': None, + 'statsPersistedAt': (now - downtime).isoformat(), + 'crawlerRuntimeMillis': previous_runtime.total_seconds() * 1000, + }, + ) + + async with Statistics.with_default_state(persistence_enabled=True, persist_state_key=key) as statistics: + runtime = statistics.state.crawler_runtime + + assert previous_runtime <= runtime < previous_runtime + timedelta(minutes=1) + + +async def test_runtime_counts_after_resume_on_host_with_clock_behind() -> None: + """A resumed run is measured even when the persist timestamp of the previous host lies in the future.""" + key = 'statistics_clock_behind' + run_in_progress_duration = timedelta(seconds=10) + + kvs = await KeyValueStore.open() + await kvs.set_value( + key, + { + 'statsPersistedAt': (datetime.now(UTC) + timedelta(hours=1)).isoformat(), + 'crawlerRuntimeMillis': 0, + }, + ) + + async with Statistics.with_default_state(persistence_enabled=True, persist_state_key=key) as statistics: + assert statistics.state.crawler_last_started_at + statistics.state.crawler_last_started_at -= run_in_progress_duration + runtime = statistics.state.crawler_runtime + + assert runtime >= run_in_progress_duration + + +async def test_serialized_state_includes_runtime_of_run_in_progress() -> None: + """A state serialized during a run carries the runtime up to that moment, so every persist is a checkpoint.""" + run_in_progress_duration = timedelta(hours=1) + + async with Statistics.with_default_state() as statistics: + # Simulate a run that has been in progress for an hour by shifting its start timestamp into the past. + assert statistics.state.crawler_last_started_at + statistics.state.crawler_last_started_at -= run_in_progress_duration + serialized = statistics.state.model_dump(by_alias=True) + + assert serialized['crawlerRuntimeMillis'] >= run_in_progress_duration.total_seconds() * 1000 + + +async def test_backwards_clock_jump_does_not_regress_runtime() -> None: + """A backwards wall clock jump during a run does not push the runtime below the accumulated total.""" + async with Statistics.with_default_state() as statistics: + # Simulate a backwards clock jump by shifting the start timestamp of the run into the future. + assert statistics.state.crawler_last_started_at + statistics.state.crawler_last_started_at += timedelta(hours=1) + runtime = statistics.state.crawler_runtime + + assert runtime == timedelta() + + +async def test_runtime_stops_growing_after_run_finishes() -> None: + """The runtime of a finished run stays fixed as time passes.""" + async with Statistics.with_default_state() as statistics: + pass + + runtime_after_exit = statistics.state.crawler_runtime + await asyncio.sleep(0.05) + + assert statistics.state.crawler_runtime == runtime_after_exit + + +async def test_runtime_carries_over_between_runs() -> None: + """A second run continues accumulating the runtime from the total persisted by the first run.""" + key = 'statistics_restored_runtime' + + async with Statistics.with_default_state(persistence_enabled=True, persist_state_key=key): + pass + + kvs = await KeyValueStore.open() + stored_state = await kvs.get_value(key) + previous_runtime = timedelta(milliseconds=stored_state['crawlerRuntimeMillis']) + + async with Statistics.with_default_state(persistence_enabled=True, persist_state_key=key) as statistics: + pass + + assert previous_runtime <= statistics.state.crawler_runtime < previous_runtime + timedelta(minutes=1) + + +async def test_persisted_finished_state_validates_to_same_runtime() -> None: + """A state persisted after a finished run reports the same runtime when validated back into the model.""" + async with Statistics.with_default_state() as statistics: + assert statistics.state.crawler_last_started_at + statistics.state.crawler_last_started_at -= timedelta(seconds=10) + + serialized = statistics.state.model_dump(mode='json', by_alias=True) + restored = StatisticsState.model_validate(serialized) + + assert abs(restored.crawler_runtime - statistics.state.crawler_runtime) < timedelta(milliseconds=1) + + +async def test_runtime_restored_from_v1_iso_duration() -> None: + """A runtime persisted by crawlee v1 as an ISO 8601 duration string is restored on resume.""" + key = 'statistics_v1_runtime' + previous_runtime = timedelta(seconds=10) + + kvs = await KeyValueStore.open() + await kvs.set_value(key, {'crawlerRuntimeMillis': 'PT10S'}) + + async with Statistics.with_default_state(persistence_enabled=True, persist_state_key=key) as statistics: + runtime = statistics.state.crawler_runtime + + assert previous_runtime <= runtime < previous_runtime + timedelta(minutes=1) + + @pytest.mark.parametrize( ('second_id', 'expected_requests_failed'), [