diff --git a/ldclient/impl/datasourcev2/async_streaming.py b/ldclient/impl/datasourcev2/async_streaming.py index 36e7d09a..0f8428ab 100644 --- a/ldclient/impl/datasourcev2/async_streaming.py +++ b/ldclient/impl/datasourcev2/async_streaming.py @@ -24,7 +24,13 @@ with_fallback_signal ) from ldclient.impl.datasystem import DiagnosticAccumulator, DiagnosticSource -from ldclient.impl.util import _LD_ENVID_HEADER, _LD_FD_FALLBACK_HEADER, log +from ldclient.impl.util import ( + _LD_ENVID_HEADER, + _LD_FD_FALLBACK_HEADER, + current_time_millis, + log, + monotonic_seconds +) from ldclient.interfaces import ( AsyncSynchronizer, ChangeSetBuilder, @@ -150,7 +156,7 @@ async def sync(self, ss: SelectorStore) -> AsyncGenerator[Update, None]: change_set_builder = ChangeSetBuilder() self._running = True - self._connection_attempt_start_time = time() + self._connection_attempt_start_time = monotonic_seconds() envid = None # A Start action with the X-LD-FD-Fallback: true header sets fallback_requested. @@ -253,9 +259,9 @@ async def _close_owned_session(self): self._owned_session = None def _record_stream_init(self, failed: bool): - if self._diagnostic_accumulator and self._connection_attempt_start_time: - current_time = int(time() * 1000) - elapsed = current_time - int(self._connection_attempt_start_time * 1000) + if self._diagnostic_accumulator and self._connection_attempt_start_time is not None: + current_time = current_time_millis() + elapsed = int(monotonic_seconds() * 1000) - int(self._connection_attempt_start_time * 1000) self._diagnostic_accumulator.record_stream_init(current_time, elapsed if elapsed >= 0 else 0, failed) async def _handle_error(self, error: Exception, envid: Optional[str]) -> Tuple[Optional[Update], bool]: diff --git a/ldclient/impl/datasourcev2/streaming.py b/ldclient/impl/datasourcev2/streaming.py index 16e533c1..5e9ab99a 100644 --- a/ldclient/impl/datasourcev2/streaming.py +++ b/ldclient/impl/datasourcev2/streaming.py @@ -28,7 +28,13 @@ ) from ldclient.impl.datasystem import DiagnosticAccumulator, DiagnosticSource from ldclient.impl.http import HTTPFactory, _base_headers -from ldclient.impl.util import _LD_ENVID_HEADER, _LD_FD_FALLBACK_HEADER, log +from ldclient.impl.util import ( + _LD_ENVID_HEADER, + _LD_FD_FALLBACK_HEADER, + current_time_millis, + log, + monotonic_seconds +) from ldclient.interfaces import ( ChangeSetBuilder, DataSourceErrorInfo, @@ -158,7 +164,7 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, None]: change_set_builder = ChangeSetBuilder() self._running = True - self._connection_attempt_start_time = time() + self._connection_attempt_start_time = monotonic_seconds() envid = None # fallback_requested is set when a Start action carries @@ -252,9 +258,9 @@ def stop(self): self._sse.close() def _record_stream_init(self, failed: bool): - if self._diagnostic_accumulator and self._connection_attempt_start_time: - current_time = int(time() * 1000) - elapsed = current_time - int(self._connection_attempt_start_time * 1000) + if self._diagnostic_accumulator and self._connection_attempt_start_time is not None: + current_time = current_time_millis() + elapsed = int(monotonic_seconds() * 1000) - int(self._connection_attempt_start_time * 1000) self._diagnostic_accumulator.record_stream_init(current_time, elapsed if elapsed >= 0 else 0, failed) def _handle_error(self, error: Exception, envid: Optional[str]) -> Tuple[Optional[Update], bool]: diff --git a/ldclient/impl/datasourcev2/streaming_common.py b/ldclient/impl/datasourcev2/streaming_common.py index cb0428d8..36e81733 100644 --- a/ldclient/impl/datasourcev2/streaming_common.py +++ b/ldclient/impl/datasourcev2/streaming_common.py @@ -23,7 +23,8 @@ _LD_FD_FALLBACK_HEADER, http_error_message, is_http_error_recoverable, - log + log, + monotonic_seconds ) from ldclient.interfaces import ( ChangeSetBuilder, @@ -167,11 +168,11 @@ def classify_stream_error( update=update, should_continue=True, should_stop=False, - next_attempt_start=time() + next_retry_delay, + next_attempt_start=monotonic_seconds() + next_retry_delay, ) if isinstance(error, HTTPStatusError): - next_attempt_start: Optional[float] = time() + next_retry_delay + next_attempt_start: Optional[float] = monotonic_seconds() + next_retry_delay error_info = DataSourceErrorInfo( DataSourceErrorKind.ERROR_RESPONSE, @@ -244,5 +245,5 @@ def classify_stream_error( update=update, should_continue=True, should_stop=False, - next_attempt_start=time() + next_retry_delay, + next_attempt_start=monotonic_seconds() + next_retry_delay, ) diff --git a/ldclient/migrations/async_migrator.py b/ldclient/migrations/async_migrator.py index 523d6cb9..3498bdea 100644 --- a/ldclient/migrations/async_migrator.py +++ b/ldclient/migrations/async_migrator.py @@ -2,7 +2,7 @@ import asyncio from abc import ABC, abstractmethod -from datetime import datetime +from datetime import timedelta from random import Random from typing import ( TYPE_CHECKING, @@ -15,7 +15,7 @@ ) from ldclient.impl.sampler import Sampler -from ldclient.impl.util import Result, log +from ldclient.impl.util import Result, log, monotonic_seconds from ldclient.migrations.tracker import OpTracker from ldclient.migrations.types import ( ExecutionOrder, @@ -419,7 +419,7 @@ async def run(self) -> OperationResult: """ Execute the configured operation and track any available measurements. """ - start = datetime.now() + start = monotonic_seconds() try: result = await self.__fn(self.__payload) @@ -428,7 +428,7 @@ async def run(self) -> OperationResult: # Record required tracker measurements if self.__measure_latency: - self.__tracker.latency(self.__origin, datetime.now() - start) + self.__tracker.latency(self.__origin, timedelta(seconds=monotonic_seconds() - start)) if self.__measure_errors and not result.is_success(): self.__tracker.error(self.__origin) diff --git a/ldclient/migrations/migrator.py b/ldclient/migrations/migrator.py index fde023f0..000b0680 100644 --- a/ldclient/migrations/migrator.py +++ b/ldclient/migrations/migrator.py @@ -2,12 +2,12 @@ import concurrent.futures from abc import ABC, abstractmethod -from datetime import datetime +from datetime import timedelta from random import Random from typing import TYPE_CHECKING, Any, Optional, Tuple, Union from ldclient.impl.sampler import Sampler -from ldclient.impl.util import Result +from ldclient.impl.util import Result, monotonic_seconds from ldclient.migrations.tracker import OpTracker from ldclient.migrations.types import ( ExecutionOrder, @@ -313,7 +313,7 @@ def run(self) -> OperationResult: """ Execute the configured operation and track any available measurements. """ - start = datetime.now() + start = monotonic_seconds() try: result = self.__fn(self.__payload) @@ -322,7 +322,7 @@ def run(self) -> OperationResult: # Record required tracker measurements if self.__measure_latency: - self.__tracker.latency(self.__origin, datetime.now() - start) + self.__tracker.latency(self.__origin, timedelta(seconds=monotonic_seconds() - start)) if self.__measure_errors and not result.is_success(): self.__tracker.error(self.__origin) diff --git a/ldclient/testing/impl/datasourcev2/test_stream_init_timing.py b/ldclient/testing/impl/datasourcev2/test_stream_init_timing.py new file mode 100644 index 00000000..6cdfc426 --- /dev/null +++ b/ldclient/testing/impl/datasourcev2/test_stream_init_timing.py @@ -0,0 +1,82 @@ +""" +Stream-init timing is measured on the monotonic clock while the reported +timestamp stays on the wall clock. These tests exercise ``_record_stream_init`` +in isolation so the clock source is observable without driving a live +connection. They assert the clock behavior only; the recording semantics +(projected baseline, negative-elapsed clamp, skip-when-cleared) are unchanged +from the wall-clock implementation. +""" + +from unittest.mock import Mock + +import pytest + +from ldclient.impl.datasourcev2 import async_streaming, streaming +from ldclient.impl.datasourcev2.async_streaming import AsyncStreamingDataSource +from ldclient.impl.datasourcev2.streaming import StreamingDataSource + +_CASES = [ + pytest.param(streaming, StreamingDataSource, id="sync"), + pytest.param(async_streaming, AsyncStreamingDataSource, id="async"), +] + + +def _make_source(cls, baseline): + # Bypass __init__ so the test depends only on the two attributes + # _record_stream_init reads. + source = cls.__new__(cls) + source._diagnostic_accumulator = Mock() + source._connection_attempt_start_time = baseline + return source + + +@pytest.mark.parametrize("module, cls", _CASES) +def test_duration_measured_on_monotonic_clock_timestamp_on_wall_clock(monkeypatch, module, cls): + monotonic = {"t": 100.0} + wall_millis = {"t": 5_000_000} + monkeypatch.setattr(module, "monotonic_seconds", lambda: monotonic["t"]) + monkeypatch.setattr(module, "current_time_millis", lambda: wall_millis["t"]) + + source = _make_source(cls, monotonic["t"]) + + # The monotonic clock advances two seconds; the wall clock jumps backward an + # hour (an NTP correction). The duration must reflect the monotonic advance, + # and the reported timestamp must reflect the (jumped) wall clock. + monotonic["t"] += 2.0 + wall_millis["t"] -= 3_600_000 + + source._record_stream_init(False) + + source._diagnostic_accumulator.record_stream_init.assert_called_once_with( + wall_millis["t"], 2000, False + ) + + +@pytest.mark.parametrize("module, cls", _CASES) +def test_negative_elapsed_is_clamped_to_zero(monkeypatch, module, cls): + monotonic = {"t": 100.0} + wall_millis = {"t": 5_000_000} + monkeypatch.setattr(module, "monotonic_seconds", lambda: monotonic["t"]) + monkeypatch.setattr(module, "current_time_millis", lambda: wall_millis["t"]) + + # The baseline is a projected future attempt start, so the measured elapsed + # is negative and must clamp to zero (unchanged from the wall-clock code). + source = _make_source(cls, monotonic["t"] + 3.0) + + source._record_stream_init(True) + + source._diagnostic_accumulator.record_stream_init.assert_called_once_with( + wall_millis["t"], 0, True + ) + + +@pytest.mark.parametrize("module, cls", _CASES) +def test_nothing_recorded_when_baseline_is_cleared(monkeypatch, module, cls): + monkeypatch.setattr(module, "monotonic_seconds", lambda: 100.0) + monkeypatch.setattr(module, "current_time_millis", lambda: 5_000_000) + + source = _make_source(cls, None) + + source._record_stream_init(False) + + source._diagnostic_accumulator.record_stream_init.assert_not_called() diff --git a/ldclient/testing/migrations/test_async_migrator.py b/ldclient/testing/migrations/test_async_migrator.py index 21cb9f94..96dc14e2 100644 --- a/ldclient/testing/migrations/test_async_migrator.py +++ b/ldclient/testing/migrations/test_async_migrator.py @@ -291,7 +291,11 @@ async def delay(payload): assert len(origins) == len(event.latencies) for o in origins: assert o in event.latencies - assert event.latencies[o] >= timedelta(milliseconds=100) + # The callable sleeps 0.1s; assert a substantial measured latency + # with margin for monotonic-clock granularity, which is coarse on + # some platforms (e.g. Windows before Python 3.13, ~16ms) and can + # under-measure a 100ms sleep. + assert event.latencies[o] >= timedelta(milliseconds=50) @pytest.mark.parametrize( "stage,origins", @@ -322,7 +326,11 @@ async def delay(payload): assert len(origins) == len(event.latencies) for o in origins: assert o in event.latencies - assert event.latencies[o] >= timedelta(milliseconds=100) + # The callable sleeps 0.1s; assert a substantial measured latency + # with margin for monotonic-clock granularity, which is coarse on + # some platforms (e.g. Windows before Python 3.13, ~16ms) and can + # under-measure a 100ms sleep. + assert event.latencies[o] >= timedelta(milliseconds=50) @pytest.mark.asyncio diff --git a/ldclient/testing/migrations/test_executor_timing.py b/ldclient/testing/migrations/test_executor_timing.py new file mode 100644 index 00000000..600372bf --- /dev/null +++ b/ldclient/testing/migrations/test_executor_timing.py @@ -0,0 +1,54 @@ +""" +Tests that migration operation latency is measured on the monotonic clock. +""" + +from datetime import timedelta +from unittest.mock import MagicMock + +import pytest + +from ldclient.impl.util import Result +from ldclient.migrations import async_migrator as async_migrator_module +from ldclient.migrations import migrator as migrator_module +from ldclient.migrations.types import Origin + + +class _FakeMonotonic: + def __init__(self, now: float = 1000.0): + self.now = now + + def __call__(self) -> float: + return self.now + + +def test_latency_is_measured_on_the_monotonic_clock(monkeypatch): + clock = _FakeMonotonic() + monkeypatch.setattr(migrator_module, "monotonic_seconds", clock) + tracker = MagicMock() + + def fn(payload): + # Simulates 42s of work; a wall-clock jump during the operation used to + # corrupt the reported latency (datetime.now() is DST-bearing local time). + clock.now = 1042.0 + return Result.success(None) + + executor = migrator_module.Executor(Origin.OLD, fn, tracker, True, False, None) + executor.run() + + tracker.latency.assert_called_once_with(Origin.OLD, timedelta(seconds=42.0)) + + +@pytest.mark.asyncio +async def test_async_latency_is_measured_on_the_monotonic_clock(monkeypatch): + clock = _FakeMonotonic() + monkeypatch.setattr(async_migrator_module, "monotonic_seconds", clock) + tracker = MagicMock() + + async def fn(payload): + clock.now = 1042.0 + return Result.success(None) + + executor = async_migrator_module.AsyncExecutor(Origin.OLD, fn, tracker, True, False, None) + await executor.run() + + tracker.latency.assert_called_once_with(Origin.OLD, timedelta(seconds=42.0)) diff --git a/ldclient/testing/migrations/test_migrator.py b/ldclient/testing/migrations/test_migrator.py index 92e26009..e82b5c55 100644 --- a/ldclient/testing/migrations/test_migrator.py +++ b/ldclient/testing/migrations/test_migrator.py @@ -204,7 +204,11 @@ def delay(payload): assert len(origins) == len(event.latencies) for o in origins: assert o in event.latencies - assert event.latencies[o] >= timedelta(milliseconds=100) + # The callable sleeps 0.1s; assert a substantial measured latency + # with margin for monotonic-clock granularity, which is coarse on + # some platforms (e.g. Windows before Python 3.13, ~16ms) and can + # under-measure a 100ms sleep. + assert event.latencies[o] >= timedelta(milliseconds=50) @pytest.mark.parametrize( "stage,origins", @@ -238,7 +242,11 @@ def delay(payload): assert len(origins) == len(event.latencies) for o in origins: assert o in event.latencies - assert event.latencies[o] >= timedelta(milliseconds=100) + # The callable sleeps 0.1s; assert a substantial measured latency + # with margin for monotonic-clock granularity, which is coarse on + # some platforms (e.g. Windows before Python 3.13, ~16ms) and can + # under-measure a 100ms sleep. + assert event.latencies[o] >= timedelta(milliseconds=50) class TestTrackingErrors: