From fbab2a801334539e804a4738539a2e152d4b4c9b Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 20 Oct 2026 07:45:07 +0000 Subject: [PATCH 1/7] fix!: do not count downtime towards crawler_runtime after migration or resurrection The `crawler_runtime` statistic no longer counts the downtime between the runs of a resumed crawler, and it accumulates correctly across any number of persist/resume cycles: - The first periodic statistics log reports exactly the runtime accumulated so far: zero for a fresh start, the previously persisted total for a resumed run. - The accumulated runtime is stored in the new `accumulated_runtime` field, persisted under `crawlerRuntimeMillis`. Every persist writes the live total, so a state persisted mid-run (migration, abort) carries the runtime up to that moment and a resumed run continues from the last checkpoint. - The timestamps of the previous run are cleared before the new run starts being measured, so they cannot leak the downtime into the runtime, and a backwards clock jump during a run cannot push the runtime below the accumulated total. BREAKING CHANGE: `StatisticsState.crawler_runtime_for_serialization` is removed; the `crawler_runtime` property is backed by the new `accumulated_runtime` field. `crawlerRuntimeMillis` is serialized as a number of milliseconds instead of an ISO 8601 duration string. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01VMHwDx87jV4rnu5LkbpNxy --- docs/upgrading/upgrading_to_v2.md | 4 + src/crawlee/statistics/_models.py | 46 ++++-- src/crawlee/statistics/_statistics.py | 14 +- tests/unit/_statistics/test_persistence.py | 161 +++++++++++++++++++++ 4 files changed, 203 insertions(+), 22 deletions(-) diff --git a/docs/upgrading/upgrading_to_v2.md b/docs/upgrading/upgrading_to_v2.md index 4bfa59ae51..c2d5652fec 100644 --- a/docs/upgrading/upgrading_to_v2.md +++ b/docs/upgrading/upgrading_to_v2.md @@ -17,6 +17,10 @@ We drop support for Python 3.10, which reached its end of life in October 2026. `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_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. + ## `RequestQueueClient.is_finished` is abstract `RequestQueueClient.is_finished` is now an abstract method. In v1, it fell back to `is_empty`, which ignores requests still being processed and could report the queue as finished too early. A subclass that doesn't implement it raises `TypeError` on instantiation. If you have a custom request queue client, implement `is_finished` so that it returns `True` only when the queue is empty and no requests are in progress. The built-in clients already implement it. diff --git a/src/crawlee/statistics/_models.py b/src/crawlee/statistics/_models.py index ede4037859..bc41f5d387 100644 --- a/src/crawlee/statistics/_models.py +++ b/src/crawlee/statistics/_models.py @@ -5,7 +5,15 @@ 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 @@ -100,25 +108,31 @@ 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() + accumulated_runtime: Annotated[timedelta_ms, Field(alias='crawlerRuntimeMillis')] = timedelta() + """The runtime accumulated by the previous runs of the crawler, excluding the run in progress. - def model_post_init(self, /, __context: Any) -> None: - self._runtime_offset = self.crawler_runtime or self._runtime_offset + When serialized, the runtime of the run in progress is included, so that a persisted state always carries + the total runtime accumulated so far. Do not assign to it: it is restored from the persisted state, and + serialization derives the live total from the timestamps. + """ @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, excluding the downtime between them. + + The run in progress is measured by the `crawler_last_started_at` and `crawler_finished_at` timestamps. + In a loaded state, the timestamps describe the previous run, whose runtime `accumulated_runtime` already + contains, so they must be reset before a new run is measured. `Statistics` resets them on enter, before + the runtime is first read. + """ + if self.crawler_last_started_at is None: + 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) -> float: + 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 4dccaa8ba7..cd31483c27 100644 --- a/src/crawlee/statistics/_statistics.py +++ b/src/crawlee/statistics/_statistics.py @@ -169,10 +169,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 and must not measure this one: left in place, + # they would count the downtime between the runs (e.g. after a migration or resurrection) towards + # the runtime. Clear them, so that the initial periodic log reports exactly the runtime accumulated + # by the previous runs (zero for a fresh start). + self.state.crawler_last_started_at = None self.state.crawler_finished_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 @@ -195,10 +200,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 3c6e06d02d..efa033b5e7 100644 --- a/tests/unit/_statistics/test_persistence.py +++ b/tests/unit/_statistics/test_persistence.py @@ -1,6 +1,15 @@ from __future__ import annotations +import asyncio +import logging +from datetime import UTC, datetime, timedelta +from typing import TYPE_CHECKING + from crawlee.statistics import Statistics +from crawlee.storages import KeyValueStore + +if TYPE_CHECKING: + import pytest async def test_basic_persistence() -> None: @@ -13,3 +22,155 @@ async def test_basic_persistence() -> None: pass 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_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 measured by a finished run stays frozen instead of following the clock.""" + 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) From fc873682f77c4c1e9cb9802fbbb66c97c626e407 Mon Sep 17 00:00:00 2001 From: Josef Prochazka Date: Wed, 21 Oct 2026 09:01:12 +0200 Subject: [PATCH 2/7] Update comments --- src/crawlee/statistics/_models.py | 16 +++------------- 1 file changed, 3 insertions(+), 13 deletions(-) diff --git a/src/crawlee/statistics/_models.py b/src/crawlee/statistics/_models.py index bc41f5d387..a731c4c31d 100644 --- a/src/crawlee/statistics/_models.py +++ b/src/crawlee/statistics/_models.py @@ -109,22 +109,11 @@ class StatisticsState(BaseModel): ] = {} accumulated_runtime: Annotated[timedelta_ms, Field(alias='crawlerRuntimeMillis')] = timedelta() - """The runtime accumulated by the previous runs of the crawler, excluding the run in progress. - - When serialized, the runtime of the run in progress is included, so that a persisted state always carries - the total runtime accumulated so far. Do not assign to it: it is restored from the persisted state, and - serialization derives the live total from the timestamps. - """ + """The runtime accumulated by the previous runs of the crawler, excluding the run in progress.""" @property def crawler_runtime(self) -> timedelta: - """The total runtime of the crawler, accumulated over all its runs, excluding the downtime between them. - - The run in progress is measured by the `crawler_last_started_at` and `crawler_finished_at` timestamps. - In a loaded state, the timestamps describe the previous run, whose runtime `accumulated_runtime` already - contains, so they must be reset before a new run is measured. `Statistics` resets them on enter, before - the runtime is first read. - """ + """The total runtime of the crawler, accumulated over all its runs.""" if self.crawler_last_started_at is None: return self.accumulated_runtime finished_at = self.crawler_finished_at or datetime.now(UTC) @@ -132,6 +121,7 @@ def crawler_runtime(self) -> timedelta: @field_serializer('accumulated_runtime') def _serialize_accumulated_runtime(self, _value: timedelta) -> float: + """Serialize the total runtime. Previously accumulated and current runtime.""" return round(self.crawler_runtime.total_seconds() * 1000) @computed_field(alias='requestTotalDurationMillis', return_type=timedelta_ms) From 41161cd2d3aa0e08caf6d9bb255f756a61f4a36b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Josef=20Proch=C3=A1zka?= Date: Thu, 29 Oct 2026 12:00:14 +0100 Subject: [PATCH 3/7] Apply batched suggestions from code review Co-authored-by: Vlada Dusek --- src/crawlee/statistics/_models.py | 14 +++-------- src/crawlee/statistics/_statistics.py | 6 ++--- tests/unit/_statistics/test_persistence.py | 28 +++++++++++++++++++++- 3 files changed, 32 insertions(+), 16 deletions(-) diff --git a/src/crawlee/statistics/_models.py b/src/crawlee/statistics/_models.py index 998898ccec..f51bb5a3ce 100644 --- a/src/crawlee/statistics/_models.py +++ b/src/crawlee/statistics/_models.py @@ -5,15 +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, - field_serializer, -) +from pydantic import BaseModel, ConfigDict, Field, PlainSerializer, PlainValidator, computed_field, field_serializer from pydantic.alias_generators import to_camel from typing_extensions import override @@ -121,8 +113,8 @@ def crawler_runtime(self) -> timedelta: return self.accumulated_runtime + max(timedelta(), finished_at - self.crawler_last_started_at) @field_serializer('accumulated_runtime') - def _serialize_accumulated_runtime(self, _value: timedelta) -> float: - """Serialize the total runtime. Previously accumulated and current 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) diff --git a/src/crawlee/statistics/_statistics.py b/src/crawlee/statistics/_statistics.py index 3c277b7b2a..eb8cd3a884 100644 --- a/src/crawlee/statistics/_statistics.py +++ b/src/crawlee/statistics/_statistics.py @@ -174,10 +174,8 @@ async def __aenter__(self) -> Self: await self._state.initialize() - # The restored timestamps belong to the previous run and must not measure this one: left in place, - # they would count the downtime between the runs (e.g. after a migration or resurrection) towards - # the runtime. Clear them, so that the initial periodic log reports exactly the runtime accumulated - # by the previous runs (zero for a fresh start). + # 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. self.state.crawler_last_started_at = None self.state.crawler_finished_at = None diff --git a/tests/unit/_statistics/test_persistence.py b/tests/unit/_statistics/test_persistence.py index 6cb837f5d1..b844f6091b 100644 --- a/tests/unit/_statistics/test_persistence.py +++ b/tests/unit/_statistics/test_persistence.py @@ -147,7 +147,7 @@ async def test_backwards_clock_jump_does_not_regress_runtime() -> None: async def test_runtime_stops_growing_after_run_finishes() -> None: - """The runtime measured by a finished run stays frozen instead of following the clock.""" + """The runtime of a finished run stays fixed as time passes.""" async with Statistics.with_default_state() as statistics: pass @@ -174,6 +174,32 @@ async def test_runtime_carries_over_between_runs() -> None: 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'), [ From d5eb28402f5beda50421bd637ed99eab3ab5d38c Mon Sep 17 00:00:00 2001 From: Josef Prochazka Date: Thu, 29 Oct 2026 13:12:08 +0100 Subject: [PATCH 4/7] Do not count runtime twice in rare cases when `StatisticsState` is used manually. --- src/crawlee/statistics/_models.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/src/crawlee/statistics/_models.py b/src/crawlee/statistics/_models.py index f51bb5a3ce..796a053a3a 100644 --- a/src/crawlee/statistics/_models.py +++ b/src/crawlee/statistics/_models.py @@ -108,7 +108,12 @@ class StatisticsState(BaseModel): def crawler_runtime(self) -> timedelta: """The total runtime of the crawler, accumulated over all its runs.""" if self.crawler_last_started_at is None: + # Crawler is not running. return self.accumulated_runtime + if self.stats_persisted_at is not None and self.crawler_last_started_at <= self.stats_persisted_at: + # Accumulated runtime before persistence is already included, 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) From e04085a39934a8a902be087128a63c684d401f39 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Josef=20Proch=C3=A1zka?= Date: Thu, 29 Oct 2026 13:15:14 +0100 Subject: [PATCH 5/7] Update docs/upgrading/upgrading_to_v2.md Co-authored-by: Vlada Dusek --- docs/upgrading/upgrading_to_v2.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/upgrading/upgrading_to_v2.md b/docs/upgrading/upgrading_to_v2.md index 45c9bbc5fe..609afc2407 100644 --- a/docs/upgrading/upgrading_to_v2.md +++ b/docs/upgrading/upgrading_to_v2.md @@ -15,7 +15,7 @@ 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 From 63caadb1899330803651dd8bda3798f67c9e0266 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Josef=20Proch=C3=A1zka?= Date: Thu, 29 Oct 2026 13:15:45 +0100 Subject: [PATCH 6/7] Update docs/upgrading/upgrading_to_v2.md Co-authored-by: Vlada Dusek --- docs/upgrading/upgrading_to_v2.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/upgrading/upgrading_to_v2.md b/docs/upgrading/upgrading_to_v2.md index 609afc2407..090bfc7c95 100644 --- a/docs/upgrading/upgrading_to_v2.md +++ b/docs/upgrading/upgrading_to_v2.md @@ -19,7 +19,7 @@ We drop support for Python 3.10, which reached its end of life in October 2026. ## `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. +`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 From 17b87cd12ce5de807f190976df38080ebea46332 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Josef=20Proch=C3=A1zka?= Date: Fri, 30 Oct 2026 08:37:26 +0100 Subject: [PATCH 7/7] Apply batched suggestions from code review Co-authored-by: Vlada Dusek --- src/crawlee/statistics/_models.py | 3 +-- src/crawlee/statistics/_statistics.py | 4 +++- tests/unit/_statistics/test_persistence.py | 22 ++++++++++++++++++++++ 3 files changed, 26 insertions(+), 3 deletions(-) diff --git a/src/crawlee/statistics/_models.py b/src/crawlee/statistics/_models.py index 796a053a3a..8a7d2a3873 100644 --- a/src/crawlee/statistics/_models.py +++ b/src/crawlee/statistics/_models.py @@ -108,10 +108,9 @@ class StatisticsState(BaseModel): def crawler_runtime(self) -> timedelta: """The total runtime of the crawler, accumulated over all its runs.""" if self.crawler_last_started_at is None: - # Crawler is not running. return self.accumulated_runtime if self.stats_persisted_at is not None and self.crawler_last_started_at <= self.stats_persisted_at: - # Accumulated runtime before persistence is already included, do not count it twice. + # 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) diff --git a/src/crawlee/statistics/_statistics.py b/src/crawlee/statistics/_statistics.py index eb8cd3a884..02c1e4e90a 100644 --- a/src/crawlee/statistics/_statistics.py +++ b/src/crawlee/statistics/_statistics.py @@ -175,9 +175,11 @@ async def __aenter__(self) -> Self: await self._state.initialize() # 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. + # 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 the restored state before the new run starts being measured. self._periodic_logger.start() diff --git a/tests/unit/_statistics/test_persistence.py b/tests/unit/_statistics/test_persistence.py index b844f6091b..696a7a25f0 100644 --- a/tests/unit/_statistics/test_persistence.py +++ b/tests/unit/_statistics/test_persistence.py @@ -122,6 +122,28 @@ async def test_runtime_after_unclean_shutdown_excludes_downtime() -> None: 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)