fix: Use monotonic clocks for migration latency and FDv2 stream-init timing

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]:

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]:

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

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)

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)

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()

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

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))

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: