fix!: exclude downtime from crawler_runtime after a resume

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
 

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

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

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'),
     [