From 4cef86b1849bde368138fbbdfe7b85ea8992f4c0 Mon Sep 17 00:00:00 2001 From: Todd Anderson Date: Fri, 16 Oct 2026 11:45:19 -0400 Subject: [PATCH 1/5] fix: Use monotonic clocks for migration latency and FDv2 stream-init timing Wall-clock steps (DST transitions via naive datetime.now(), NTP corrections, VM/container time resync) could report wildly wrong or negative migration-operation latencies, and could inflate or zero out the FDv2 stream-init durations reported in diagnostic events. Migration latency (sync and async) is now measured with a monotonic clock and delivered to the tracker as a timedelta, unchanged in type. FDv2 stream-init duration is measured monotonically from the real connection moment (the SSE Start action) rather than a wall-clock projection of the next attempt; the wall clock is kept only for the reported timestamp. The projected next-attempt baseline is removed, and the diagnostic accumulator now clamps durationMillis to a non-negative integer once for every caller. Teardown paths (clean close and unexpected event errors) clear the attempt baseline so a later reconnect failure cannot report a duration spanning the backoff sleep. SDK-3307 --- ldclient/impl/datasourcev2/async_streaming.py | 46 ++++--- ldclient/impl/datasourcev2/streaming.py | 44 ++++--- .../impl/datasourcev2/streaming_common.py | 20 +-- ldclient/impl/events/diagnostics.py | 1 + ldclient/migrations/async_migrator.py | 8 +- ldclient/migrations/migrator.py | 8 +- .../datasourcev2/test_stream_init_timing.py | 115 ++++++++++++++++++ .../test_streaming_synchronizer.py | 74 +++++++++++ .../migrations/test_executor_timing.py | 54 ++++++++ 9 files changed, 311 insertions(+), 59 deletions(-) create mode 100644 ldclient/testing/impl/datasourcev2/test_stream_init_timing.py create mode 100644 ldclient/testing/migrations/test_executor_timing.py diff --git a/ldclient/impl/datasourcev2/async_streaming.py b/ldclient/impl/datasourcev2/async_streaming.py index 36e7d09a..b2e516b8 100644 --- a/ldclient/impl/datasourcev2/async_streaming.py +++ b/ldclient/impl/datasourcev2/async_streaming.py @@ -4,7 +4,7 @@ """ import json -from time import time +import time from typing import AsyncGenerator, Callable, Optional, Tuple from urllib import parse @@ -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, @@ -117,7 +123,7 @@ def __init__( self._owned_session: Optional[aiohttp.ClientSession] = None self._running = False self._diagnostic_accumulator: Optional[DiagnosticAccumulator] = None - self._connection_attempt_start_time: Optional[float] = None + self._connection_attempt_started_monotonic: Optional[float] = None def set_diagnostic_accumulator(self, diagnostic_accumulator: DiagnosticAccumulator): self._diagnostic_accumulator = diagnostic_accumulator @@ -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_started_monotonic = monotonic_seconds() envid = None # A Start action with the X-LD-FD-Fallback: true header sets fallback_requested. @@ -167,6 +173,7 @@ async def sync(self, ss: SelectorStore) -> AsyncGenerator[Update, None]: # emit a fault with no-error. We can ignore this since we want # the connection to continue. if action.error is None: + self._connection_attempt_started_monotonic = None continue if action.headers is not None: @@ -183,10 +190,12 @@ async def sync(self, ss: SelectorStore) -> AsyncGenerator[Update, None]: break continue - if isinstance(action, Start) and action.headers is not None: - envid = action.headers.get(_LD_ENVID_HEADER, envid) - if action.headers.get(_LD_FD_FALLBACK_HEADER) == 'true': - fallback_requested = True + if isinstance(action, Start): + self._connection_attempt_started_monotonic = monotonic_seconds() + if action.headers is not None: + envid = action.headers.get(_LD_ENVID_HEADER, envid) + if action.headers.get(_LD_FD_FALLBACK_HEADER) == 'true': + fallback_requested = True if not isinstance(action, Event): continue @@ -195,7 +204,7 @@ async def sync(self, ss: SelectorStore) -> AsyncGenerator[Update, None]: update = process_message(action, change_set_builder, envid) if update is not None: self._record_stream_init(False) - self._connection_attempt_start_time = None + self._connection_attempt_started_monotonic = None if fallback_requested: # The completed update is the natural moment to honor # the latched directive: yield once with the signal, @@ -219,11 +228,12 @@ async def sync(self, ss: SelectorStore) -> AsyncGenerator[Update, None]: "Error while handling stream event; will restart stream: %s", e ) await self._sse.interrupt() + self._connection_attempt_started_monotonic = None yield with_fallback_signal(Update( state=DataSourceState.INTERRUPTED, error=DataSourceErrorInfo( - DataSourceErrorKind.UNKNOWN, 0, time(), str(e) + DataSourceErrorKind.UNKNOWN, 0, time.time(), str(e) ), fallback_to_fdv1=False, environment_id=envid, @@ -253,10 +263,12 @@ 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) - self._diagnostic_accumulator.record_stream_init(current_time, elapsed if elapsed >= 0 else 0, failed) + if self._diagnostic_accumulator and self._connection_attempt_started_monotonic is not None: + # The reported timestamp is a wall-clock value; the duration comes from + # the monotonic clock so a wall-clock step cannot distort it. + current_time = current_time_millis() + elapsed = int((monotonic_seconds() - self._connection_attempt_started_monotonic) * 1000) + self._diagnostic_accumulator.record_stream_init(current_time, elapsed, failed) async def _handle_error(self, error: Exception, envid: Optional[str]) -> Tuple[Optional[Update], bool]: """ @@ -273,11 +285,9 @@ async def _handle_error(self, error: Exception, envid: Optional[str]) -> Tuple[O if not self._running: return (None, False) # don't retry if we've been deliberately stopped - decision = classify_stream_error( - error, self._sse.next_retry_delay, envid # type: ignore - ) + decision = classify_stream_error(error, envid) self._record_stream_init(True) - self._connection_attempt_start_time = decision.next_attempt_start + self._connection_attempt_started_monotonic = None if decision.should_stop: await self.stop() diff --git a/ldclient/impl/datasourcev2/streaming.py b/ldclient/impl/datasourcev2/streaming.py index 16e533c1..f32a2630 100644 --- a/ldclient/impl/datasourcev2/streaming.py +++ b/ldclient/impl/datasourcev2/streaming.py @@ -4,7 +4,7 @@ """ import json -from time import time +import time from typing import Callable, Generator, Optional, Tuple from urllib import parse @@ -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, @@ -126,7 +132,7 @@ def __init__(self, self._sse: Optional[SSEClient] = None self._running = False self._diagnostic_accumulator: Optional[DiagnosticAccumulator] = None - self._connection_attempt_start_time: Optional[float] = None + self._connection_attempt_started_monotonic: Optional[float] = None def set_diagnostic_accumulator(self, diagnostic_accumulator: DiagnosticAccumulator): self._diagnostic_accumulator = diagnostic_accumulator @@ -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_started_monotonic = monotonic_seconds() envid = None # fallback_requested is set when a Start action carries @@ -176,6 +182,7 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, None]: # emit a fault with no-error. We can ignore this since we want # the connection to continue. if action.error is None: + self._connection_attempt_started_monotonic = None continue if action.headers is not None: @@ -192,10 +199,12 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, None]: break continue - if isinstance(action, Start) and action.headers is not None: - envid = action.headers.get(_LD_ENVID_HEADER, envid) - if action.headers.get(_LD_FD_FALLBACK_HEADER) == 'true': - fallback_requested = True + if isinstance(action, Start): + self._connection_attempt_started_monotonic = monotonic_seconds() + if action.headers is not None: + envid = action.headers.get(_LD_ENVID_HEADER, envid) + if action.headers.get(_LD_FD_FALLBACK_HEADER) == 'true': + fallback_requested = True if not isinstance(action, Event): continue @@ -204,7 +213,7 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, None]: update = process_message(action, change_set_builder, envid) if update is not None: self._record_stream_init(False) - self._connection_attempt_start_time = None + self._connection_attempt_started_monotonic = None if fallback_requested: # The completed update is the natural moment to honor # the latched directive: yield once with the signal, @@ -228,11 +237,12 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, None]: "Error while handling stream event; will restart stream: %s", e ) self._sse.interrupt() + self._connection_attempt_started_monotonic = None yield with_fallback_signal(Update( state=DataSourceState.INTERRUPTED, error=DataSourceErrorInfo( - DataSourceErrorKind.UNKNOWN, 0, time(), str(e) + DataSourceErrorKind.UNKNOWN, 0, time.time(), str(e) ), fallback_to_fdv1=False, environment_id=envid, @@ -252,10 +262,10 @@ 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) - self._diagnostic_accumulator.record_stream_init(current_time, elapsed if elapsed >= 0 else 0, failed) + if self._diagnostic_accumulator and self._connection_attempt_started_monotonic is not None: + current_time = current_time_millis() + elapsed = int((monotonic_seconds() - self._connection_attempt_started_monotonic) * 1000) + self._diagnostic_accumulator.record_stream_init(current_time, elapsed, failed) def _handle_error(self, error: Exception, envid: Optional[str]) -> Tuple[Optional[Update], bool]: """ @@ -272,11 +282,9 @@ def _handle_error(self, error: Exception, envid: Optional[str]) -> Tuple[Optiona if not self._running: return (None, False) # don't retry if we've been deliberately stopped - decision = classify_stream_error( - error, self._sse.next_retry_delay, envid # type: ignore - ) + decision = classify_stream_error(error, envid) self._record_stream_init(True) - self._connection_attempt_start_time = decision.next_attempt_start + self._connection_attempt_started_monotonic = None if decision.should_stop: self.stop() diff --git a/ldclient/impl/datasourcev2/streaming_common.py b/ldclient/impl/datasourcev2/streaming_common.py index cb0428d8..f6aca4fd 100644 --- a/ldclient/impl/datasourcev2/streaming_common.py +++ b/ldclient/impl/datasourcev2/streaming_common.py @@ -132,26 +132,23 @@ class StreamErrorDecision: The values are computed without any I/O so both the sync and async synchronizers share one decision, then apply the side effects themselves: - they record a failed stream init, set the connection attempt start time to - ``next_attempt_start``, stop the synchronizer when ``should_stop`` is set, - and retry the connection when ``should_continue`` is set. + they record a failed stream init, stop the synchronizer when ``should_stop`` + is set, and retry the connection when ``should_continue`` is set. """ update: Update should_continue: bool should_stop: bool - next_attempt_start: Optional[float] def classify_stream_error( - error: Exception, next_retry_delay: float, envid: Optional[str] + error: Exception, envid: Optional[str] ) -> StreamErrorDecision: """ Decide how to react to a streaming connection error. The caller must first check that the synchronizer is still running; this - function assumes it is. It returns a decision the caller applies. The caller - always records a failed stream init before it sets the connection attempt - start time to the returned ``next_attempt_start``. + function assumes it is. It returns a decision the caller applies, recording + a failed stream init before doing so. """ if isinstance(error, json.decoder.JSONDecodeError): log.error("Unexpected error on stream connection: %s, will retry", error) @@ -167,12 +164,9 @@ def classify_stream_error( update=update, should_continue=True, should_stop=False, - next_attempt_start=time() + next_retry_delay, ) if isinstance(error, HTTPStatusError): - next_attempt_start: Optional[float] = time() + next_retry_delay - error_info = DataSourceErrorInfo( DataSourceErrorKind.ERROR_RESPONSE, error.status, @@ -194,7 +188,6 @@ def classify_stream_error( update=update, should_continue=False, should_stop=True, - next_attempt_start=next_attempt_start, ) http_error_message_result = http_error_message( @@ -218,7 +211,6 @@ def classify_stream_error( update=update, should_continue=False, should_stop=True, - next_attempt_start=None, ) log.warning(http_error_message_result) @@ -226,7 +218,6 @@ def classify_stream_error( update=update, should_continue=True, should_stop=False, - next_attempt_start=next_attempt_start, ) log.warning("Unexpected error on stream connection: %s, will retry", error) @@ -244,5 +235,4 @@ def classify_stream_error( update=update, should_continue=True, should_stop=False, - next_attempt_start=time() + next_retry_delay, ) diff --git a/ldclient/impl/events/diagnostics.py b/ldclient/impl/events/diagnostics.py index 9a799c75..e65c127e 100644 --- a/ldclient/impl/events/diagnostics.py +++ b/ldclient/impl/events/diagnostics.py @@ -16,6 +16,7 @@ def __init__(self, diagnostic_id): self._stream_inits = [] def record_stream_init(self, timestamp, duration, failed): + duration = max(0, duration) with self._state_lock: self._stream_inits.append({'timestamp': timestamp, 'durationMillis': duration, 'failed': failed}) 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..5fda4f4d --- /dev/null +++ b/ldclient/testing/impl/datasourcev2/test_stream_init_timing.py @@ -0,0 +1,115 @@ +""" +Tests for the monotonic stream-init timing in the FDv2 streaming synchronizers. +""" + +import asyncio +import time + +from ldclient.impl.datasourcev2 import ( + async_streaming as async_streaming_module +) +from ldclient.impl.datasourcev2 import streaming as streaming_module +from ldclient.impl.datasourcev2.async_streaming import AsyncStreamingDataSource +from ldclient.impl.datasourcev2.streaming import StreamingDataSource +from ldclient.impl.events.diagnostics import _DiagnosticAccumulator + + +class _Recorder: + def __init__(self): + self.recorded = [] + + def record_stream_init(self, timestamp, duration, failed): + self.recorded.append((timestamp, duration, failed)) + + +class _FakeMonotonic: + def __init__(self, now: float = 1000.0): + self.now = now + + def __call__(self) -> float: + return self.now + + +def _bare(cls): + proc = object.__new__(cls) + recorder = _Recorder() + proc._diagnostic_accumulator = recorder + return proc, recorder + + +def test_duration_unaffected_by_forward_wall_step(monkeypatch): + # A forward wall step used to inflate the reported duration by the step size. + for cls in (StreamingDataSource, AsyncStreamingDataSource): + proc, recorder = _bare(cls) + proc._connection_attempt_started_monotonic = time.monotonic() + + real_time = time.time() + monkeypatch.setattr(time, "time", lambda: real_time + 3600) + proc._record_stream_init(False) + monkeypatch.undo() + + _, duration, failed = recorder.recorded[0] + assert failed is False + assert duration >= 0 + assert duration < 1000 + + +def test_backward_wall_step_does_not_zero_the_measurement(monkeypatch): + # A backward wall step used to drive the elapsed value negative, and the old + # clamp silently reported 0 instead of the real measurement. + for cls, module in ( + (StreamingDataSource, streaming_module), + (AsyncStreamingDataSource, async_streaming_module), + ): + proc, recorder = _bare(cls) + clock = _FakeMonotonic(1000.0) + monkeypatch.setattr(module, "monotonic_seconds", clock) + proc._connection_attempt_started_monotonic = clock() + clock.now = 1000.5 + + real_time = time.time() + monkeypatch.setattr(time, "time", lambda: real_time - 120) + proc._record_stream_init(True) + monkeypatch.undo() + + _, duration, failed = recorder.recorded[0] + assert failed is True + assert duration == 500 + + +def test_error_handling_clears_the_baseline(): + # The projection was removed: _handle_error no longer fabricates the next + # attempt's start (the library owns the reconnect sleep, so its real start + # is not visible here). It clears the baseline; the next Start re-stamps it. + proc = object.__new__(StreamingDataSource) + proc._running = True + proc._diagnostic_accumulator = None + proc._connection_attempt_started_monotonic = 1234.0 + + update, should_continue = proc._handle_error(Exception("boom"), None) + + assert should_continue is True + assert update is not None + assert proc._connection_attempt_started_monotonic is None + + +def test_async_error_handling_clears_the_baseline(): + proc = object.__new__(AsyncStreamingDataSource) + proc._running = True + proc._diagnostic_accumulator = None + proc._connection_attempt_started_monotonic = 1234.0 + + update, should_continue = asyncio.run(proc._handle_error(Exception("boom"), None)) + + assert should_continue is True + assert update is not None + assert proc._connection_attempt_started_monotonic is None + + +def test_diagnostic_accumulator_clamps_negative_durations(): + # The diagnostic schema requires durationMillis >= 0; the accumulator + # enforces it once for every call site. + accumulator = _DiagnosticAccumulator("diag-id") + accumulator.record_stream_init(123, -5, False) + + assert accumulator._stream_inits[0]['durationMillis'] == 0 diff --git a/ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py b/ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py index 216013b4..f400373a 100644 --- a/ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py +++ b/ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py @@ -3,6 +3,7 @@ import json from abc import abstractmethod +from types import SimpleNamespace from typing import Iterable, List, Optional import pytest @@ -805,3 +806,76 @@ def test_envid_missing_when_no_headers(): assert len(updates) == 1 assert updates[0].state == DataSourceState.VALID assert updates[0].environment_id is None + + +def test_stream_init_duration_measured_from_start_not_run_start(monkeypatch): + # Regression for the projected-baseline bug (finding #41): the stream-init + # duration must be measured from the real connection moment (the Start + # action), not from run() start and not from a projection in _handle_error. + from ldclient.impl.datasourcev2 import streaming as streaming_module + + server_intent = ServerIntent( + payload=Payload( + id="id", + target=300, + code=IntentCode.TRANSFER_NONE, + reason="up-to-date", + ) + ) + intent_event = Event( + event=EventName.SERVER_INTENT, + data=json.dumps(server_intent.to_dict()), + ) + + # monotonic_seconds is read at: run() start, each Start action, and + # _record_stream_init. Script the three reads so the Start stamp (105.0) is + # distinct from run-start (100.0). The recorded duration must come out as + # 105.5 - 105.0 = 0.5s; a run-start baseline would give 5.5s. + reads = iter([100.0, 105.0, 105.5]) + last = [105.5] + + def fake_monotonic(): + try: + last[0] = next(reads) + except StopIteration: + pass + return last[0] + + monkeypatch.setattr(streaming_module, "monotonic_seconds", fake_monotonic) + + recorded = [] + synchronizer = make_streaming_data_source() + synchronizer._sse_client_builder = list_sse_client([Start(), intent_event]) + synchronizer.set_diagnostic_accumulator( + SimpleNamespace(record_stream_init=lambda ts, dur, failed: recorded.append((dur, failed))) + ) + + updates = list(synchronizer.sync(MockSelectorStore(Selector.no_selector()))) + + assert len(updates) == 1 + assert recorded == [(500, False)] + + +def test_clean_close_mid_init_does_not_leave_a_stale_baseline(): + # Regression for the stale-baseline bug (#1): a connection that gets a Start + # but no completing payload, then is cleanly closed, must not leave its + # baseline set -- otherwise a later failed reconnect would record a duration + # spanning the closed connection plus the backoff sleep. Here the mid-init + # clean close + reconnect failure records nothing, rather than a stale value. + server_intent = ServerIntent( + payload=Payload(id="id", target=300, code=IntentCode.TRANSFER_FULL, reason="cant-catchup"), + ) + intent_event = Event(event=EventName.SERVER_INTENT, data=json.dumps(server_intent.to_dict())) + + recorded = [] + synchronizer = make_streaming_data_source() + synchronizer._sse_client_builder = list_sse_client([ + Start(), intent_event, Fault(None), Fault(HTTPStatusError(503)), + ]) + synchronizer.set_diagnostic_accumulator( + SimpleNamespace(record_stream_init=lambda ts, dur, failed: recorded.append((dur, failed))) + ) + + list(synchronizer.sync(MockSelectorStore(Selector.no_selector()))) + + assert recorded == [] 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)) From 56370aefd57273e1ee5b5c084a4593e14cfab0ba Mon Sep 17 00:00:00 2001 From: Todd Anderson Date: Wed, 21 Oct 2026 14:06:30 -0400 Subject: [PATCH 2/5] fix: Record one FDv2 stream-init per connection attempt The prior change stamped the stream-init baseline only on a successful connect (SSE Start), so reconnect attempts that failed to connect went unrecorded -- during a sustained outage only ~1 of N failures was reported, a regression from main which recorded every attempt. Re-stamp the baseline monotonically at each fault/teardown so every attempt is counted again, while keeping the monotonic (non-negative) durations; a reconnect-failure duration now includes the library-owned backoff, which the SDK cannot see past. Also tightens the new tests (scripted-clock clean-close assertion with teeth against a stale baseline; async healthy-drop and clean-close mirrors) and fixes the import sorting flagged by the repo's pinned isort. SDK-3307 --- ldclient/impl/datasourcev2/async_streaming.py | 6 +- ldclient/impl/datasourcev2/streaming.py | 6 +- .../impl/datasourcev2/test_async_streaming.py | 70 ++++++++++++++++ .../datasourcev2/test_stream_init_timing.py | 19 +++-- .../test_streaming_synchronizer.py | 82 +++++++++++++++++-- 5 files changed, 161 insertions(+), 22 deletions(-) diff --git a/ldclient/impl/datasourcev2/async_streaming.py b/ldclient/impl/datasourcev2/async_streaming.py index b2e516b8..e25a1daa 100644 --- a/ldclient/impl/datasourcev2/async_streaming.py +++ b/ldclient/impl/datasourcev2/async_streaming.py @@ -173,7 +173,7 @@ async def sync(self, ss: SelectorStore) -> AsyncGenerator[Update, None]: # emit a fault with no-error. We can ignore this since we want # the connection to continue. if action.error is None: - self._connection_attempt_started_monotonic = None + self._connection_attempt_started_monotonic = monotonic_seconds() continue if action.headers is not None: @@ -228,7 +228,7 @@ async def sync(self, ss: SelectorStore) -> AsyncGenerator[Update, None]: "Error while handling stream event; will restart stream: %s", e ) await self._sse.interrupt() - self._connection_attempt_started_monotonic = None + self._connection_attempt_started_monotonic = monotonic_seconds() yield with_fallback_signal(Update( state=DataSourceState.INTERRUPTED, @@ -287,7 +287,7 @@ async def _handle_error(self, error: Exception, envid: Optional[str]) -> Tuple[O decision = classify_stream_error(error, envid) self._record_stream_init(True) - self._connection_attempt_started_monotonic = None + self._connection_attempt_started_monotonic = monotonic_seconds() if decision.should_stop: await self.stop() diff --git a/ldclient/impl/datasourcev2/streaming.py b/ldclient/impl/datasourcev2/streaming.py index f32a2630..3c5dd36f 100644 --- a/ldclient/impl/datasourcev2/streaming.py +++ b/ldclient/impl/datasourcev2/streaming.py @@ -182,7 +182,7 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, None]: # emit a fault with no-error. We can ignore this since we want # the connection to continue. if action.error is None: - self._connection_attempt_started_monotonic = None + self._connection_attempt_started_monotonic = monotonic_seconds() continue if action.headers is not None: @@ -237,7 +237,7 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, None]: "Error while handling stream event; will restart stream: %s", e ) self._sse.interrupt() - self._connection_attempt_started_monotonic = None + self._connection_attempt_started_monotonic = monotonic_seconds() yield with_fallback_signal(Update( state=DataSourceState.INTERRUPTED, @@ -284,7 +284,7 @@ def _handle_error(self, error: Exception, envid: Optional[str]) -> Tuple[Optiona decision = classify_stream_error(error, envid) self._record_stream_init(True) - self._connection_attempt_started_monotonic = None + self._connection_attempt_started_monotonic = monotonic_seconds() if decision.should_stop: self.stop() diff --git a/ldclient/testing/impl/datasourcev2/test_async_streaming.py b/ldclient/testing/impl/datasourcev2/test_async_streaming.py index b3c55f02..c417975c 100644 --- a/ldclient/testing/impl/datasourcev2/test_async_streaming.py +++ b/ldclient/testing/impl/datasourcev2/test_async_streaming.py @@ -1,6 +1,7 @@ # pylint: disable=missing-docstring, too-few-public-methods import json +from types import SimpleNamespace from typing import AsyncIterable, List, Optional from unittest import mock @@ -549,3 +550,72 @@ async def test_envid_preserved_across_headerless_reconnect_start(): assert len(updates) == 2 assert updates[0].environment_id == 'my-env' assert updates[1].environment_id == 'my-env' + + +@pytest.mark.asyncio +async def test_failed_init_recorded_for_every_connect_attempt(): + # DIAG one-entry-per-attempt, exercised through the async run loop: three + # consecutive connect failures produce three failed records. + recorded = [] + src = make_streaming_data_source() + src.set_diagnostic_accumulator( + SimpleNamespace(record_stream_init=lambda ts, dur, failed: recorded.append((dur, failed))) + ) + + await collect_updates(src, [ + Fault(HTTPStatusError(503)), + Fault(HTTPStatusError(503)), + Fault(HTTPStatusError(503)), + ]) + + assert len(recorded) == 3 + assert all(failed for _, failed in recorded) + assert all(dur >= 0 for dur, _ in recorded) + + +@pytest.mark.asyncio +async def test_healthy_connection_drop_is_not_recorded_as_failed_init(): + # After a successful init, a mid-stream drop must NOT be recorded as a + # failed init (async run loop). + recorded = [] + src = make_streaming_data_source() + src.set_diagnostic_accumulator( + SimpleNamespace(record_stream_init=lambda ts, dur, failed: recorded.append((dur, failed))) + ) + + await collect_updates(src, [ + Start(), server_intent_event(IntentCode.TRANSFER_NONE), Fault(HTTPStatusError(503)), + ]) + + assert len(recorded) == 1 + assert recorded[0][1] is False + + +@pytest.mark.asyncio +async def test_reconnect_failure_after_clean_close_is_counted_without_stale_duration(monkeypatch): + # Async mirror of the sync clean-close teeth test: baseline re-stamped at the + # clean close (t=500.0), failure recorded 0.3s later -> 300ms. A stale + # baseline (left at the Start, t=200.0) would report 300300ms. + reads = iter([100.0, 200.0, 500.0, 500.3]) + last = [500.3] + + def fake_monotonic(): + try: + last[0] = next(reads) + except StopIteration: + pass + return last[0] + + monkeypatch.setattr(async_streaming, "monotonic_seconds", fake_monotonic) + + recorded = [] + src = make_streaming_data_source() + src.set_diagnostic_accumulator( + SimpleNamespace(record_stream_init=lambda ts, dur, failed: recorded.append((dur, failed))) + ) + + await collect_updates(src, [ + Start(), server_intent_event(IntentCode.TRANSFER_FULL), Fault(None), Fault(HTTPStatusError(503)), + ]) + + assert recorded == [(300, True)] diff --git a/ldclient/testing/impl/datasourcev2/test_stream_init_timing.py b/ldclient/testing/impl/datasourcev2/test_stream_init_timing.py index 5fda4f4d..c7ae27ac 100644 --- a/ldclient/testing/impl/datasourcev2/test_stream_init_timing.py +++ b/ldclient/testing/impl/datasourcev2/test_stream_init_timing.py @@ -5,9 +5,8 @@ import asyncio import time -from ldclient.impl.datasourcev2 import ( +from ldclient.impl.datasourcev2 import \ async_streaming as async_streaming_module -) from ldclient.impl.datasourcev2 import streaming as streaming_module from ldclient.impl.datasourcev2.async_streaming import AsyncStreamingDataSource from ldclient.impl.datasourcev2.streaming import StreamingDataSource @@ -77,10 +76,10 @@ def test_backward_wall_step_does_not_zero_the_measurement(monkeypatch): assert duration == 500 -def test_error_handling_clears_the_baseline(): - # The projection was removed: _handle_error no longer fabricates the next - # attempt's start (the library owns the reconnect sleep, so its real start - # is not visible here). It clears the baseline; the next Start re-stamps it. +def test_error_handling_restamps_the_baseline(): + # _handle_error records the failed init, then re-stamps the baseline to now + # so the next (library-owned) reconnect attempt is still counted. It is not + # a projection and not cleared to None. proc = object.__new__(StreamingDataSource) proc._running = True proc._diagnostic_accumulator = None @@ -90,10 +89,11 @@ def test_error_handling_clears_the_baseline(): assert should_continue is True assert update is not None - assert proc._connection_attempt_started_monotonic is None + assert proc._connection_attempt_started_monotonic is not None + assert proc._connection_attempt_started_monotonic != 1234.0 -def test_async_error_handling_clears_the_baseline(): +def test_async_error_handling_restamps_the_baseline(): proc = object.__new__(AsyncStreamingDataSource) proc._running = True proc._diagnostic_accumulator = None @@ -103,7 +103,8 @@ def test_async_error_handling_clears_the_baseline(): assert should_continue is True assert update is not None - assert proc._connection_attempt_started_monotonic is None + assert proc._connection_attempt_started_monotonic is not None + assert proc._connection_attempt_started_monotonic != 1234.0 def test_diagnostic_accumulator_clamps_negative_durations(): diff --git a/ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py b/ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py index f400373a..eeeef6b2 100644 --- a/ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py +++ b/ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py @@ -856,12 +856,80 @@ def fake_monotonic(): assert recorded == [(500, False)] -def test_clean_close_mid_init_does_not_leave_a_stale_baseline(): - # Regression for the stale-baseline bug (#1): a connection that gets a Start - # but no completing payload, then is cleanly closed, must not leave its - # baseline set -- otherwise a later failed reconnect would record a duration - # spanning the closed connection plus the backoff sleep. Here the mid-init - # clean close + reconnect failure records nothing, rather than a stale value. +def _transfer_none_intent() -> Event: + return Event( + event=EventName.SERVER_INTENT, + data=json.dumps(ServerIntent(payload=Payload( + id="id", target=300, code=IntentCode.TRANSFER_NONE, reason="up-to-date", + )).to_dict()), + ) + + +def test_failed_init_recorded_for_every_connect_attempt(): + # DIAG requires one stream-init entry per connection attempt. Three + # consecutive connect failures must produce three failed records -- the + # baseline is re-stamped after each failure so the next attempt is counted. + recorded = [] + synchronizer = make_streaming_data_source() + synchronizer._sse_client_builder = list_sse_client([ + Fault(HTTPStatusError(503)), + Fault(HTTPStatusError(503)), + Fault(HTTPStatusError(503)), + ]) + synchronizer.set_diagnostic_accumulator( + SimpleNamespace(record_stream_init=lambda ts, dur, failed: recorded.append((dur, failed))) + ) + + list(synchronizer.sync(MockSelectorStore(Selector.no_selector()))) + + assert len(recorded) == 3 + assert all(failed for _, failed in recorded) + assert all(dur >= 0 for dur, _ in recorded) + + +def test_healthy_connection_drop_is_not_recorded_as_failed_init(): + # After a successful init, a mid-stream drop must NOT be recorded as a + # failed init -- the init already succeeded, so only the success is logged. + recorded = [] + synchronizer = make_streaming_data_source() + synchronizer._sse_client_builder = list_sse_client([ + Start(), _transfer_none_intent(), Fault(HTTPStatusError(503)), + ]) + synchronizer.set_diagnostic_accumulator( + SimpleNamespace(record_stream_init=lambda ts, dur, failed: recorded.append((dur, failed))) + ) + + list(synchronizer.sync(MockSelectorStore(Selector.no_selector()))) + + assert len(recorded) == 1 + assert recorded[0][1] is False + + +def test_reconnect_failure_after_clean_close_is_counted_without_stale_duration(monkeypatch): + # A connection that gets a Start but no completing payload, then is cleanly + # closed, re-stamps rather than leaving a stale baseline. A later failed + # reconnect is still counted, and its duration is measured from the clean + # close -- never a stale value spanning the closed connection and backoff. + # + # The monotonic clock is scripted so the Start, the clean close, and the + # reconnect fault land at known times: baseline is re-stamped at the clean + # close (t=500.0) and the failure is recorded 0.3s later, so the duration is + # exactly 300ms. A stale baseline (left at the Start, t=200.0) would instead + # report 300300ms -- the exact-value assert is what gives this test teeth. + from ldclient.impl.datasourcev2 import streaming as streaming_module + + reads = iter([100.0, 200.0, 500.0, 500.3]) + last = [500.3] + + def fake_monotonic(): + try: + last[0] = next(reads) + except StopIteration: + pass + return last[0] + + monkeypatch.setattr(streaming_module, "monotonic_seconds", fake_monotonic) + server_intent = ServerIntent( payload=Payload(id="id", target=300, code=IntentCode.TRANSFER_FULL, reason="cant-catchup"), ) @@ -878,4 +946,4 @@ def test_clean_close_mid_init_does_not_leave_a_stale_baseline(): list(synchronizer.sync(MockSelectorStore(Selector.no_selector()))) - assert recorded == [] + assert recorded == [(300, True)] From 76d1c352ac631ab4630ad542afb31afeccba94a5 Mon Sep 17 00:00:00 2001 From: Todd Anderson Date: Fri, 23 Oct 2026 09:05:58 -0400 Subject: [PATCH 3/5] test: Tolerate monotonic-clock granularity in migration latency assertions After switching migration latency to a monotonic clock, the TrackingLatency tests failed on Windows before Python 3.13, where time.monotonic() has ~16ms resolution (GetTickCount64) and under-measured the 0.1s callable sleep to ~94ms. Assert a substantial measured latency with margin for clock granularity instead of >= the exact sleep duration. SDK-3307 --- ldclient/testing/migrations/test_async_migrator.py | 12 ++++++++++-- ldclient/testing/migrations/test_migrator.py | 12 ++++++++++-- 2 files changed, 20 insertions(+), 4 deletions(-) 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_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: From b5879b90dfc646dd07a8c677aa116d84184b540c Mon Sep 17 00:00:00 2001 From: Todd Anderson Date: Fri, 23 Oct 2026 13:18:00 -0400 Subject: [PATCH 4/5] refactor: Rework to a pure monotonic-clock swap (no behavioral change) The earlier revision changed stream-init recording behavior (stamping at the SSE Start event and recording one init per connection attempt). This restores main's recording structure exactly and changes only the clock source: - FDv2 stream-init duration and the next_attempt_start projection now use monotonic_seconds(); the reported stream-init timestamp and the three DataSourceErrorInfo timestamps stay on the wall clock. - The negative-elapsed clamp, the record-iff-baseline-set guard, and all Start/Fault control flow match main. - diagnostics.py reverted to main (per-site clamp retained, no central clamp). - Stream-init duration is measured as difference-of-floored-milliseconds on the monotonic clock, matching main's rounding exactly. No observable behavioral change versus main, verified by a three-agent review (general, security, adversarial) plus proof tests. --- ldclient/impl/datasourcev2/async_streaming.py | 39 +++-- ldclient/impl/datasourcev2/streaming.py | 37 ++--- .../impl/datasourcev2/streaming_common.py | 23 ++- ldclient/impl/events/diagnostics.py | 1 - .../impl/datasourcev2/test_async_streaming.py | 70 --------- .../datasourcev2/test_stream_init_timing.py | 148 +++++++----------- .../test_streaming_synchronizer.py | 142 ----------------- 7 files changed, 112 insertions(+), 348 deletions(-) diff --git a/ldclient/impl/datasourcev2/async_streaming.py b/ldclient/impl/datasourcev2/async_streaming.py index e25a1daa..f5701c6f 100644 --- a/ldclient/impl/datasourcev2/async_streaming.py +++ b/ldclient/impl/datasourcev2/async_streaming.py @@ -4,7 +4,7 @@ """ import json -import time +from time import time from typing import AsyncGenerator, Callable, Optional, Tuple from urllib import parse @@ -123,7 +123,8 @@ def __init__( self._owned_session: Optional[aiohttp.ClientSession] = None self._running = False self._diagnostic_accumulator: Optional[DiagnosticAccumulator] = None - self._connection_attempt_started_monotonic: Optional[float] = None + # Monotonic-clock baseline for the current stream-init duration. + self._connection_attempt_start_time: Optional[float] = None def set_diagnostic_accumulator(self, diagnostic_accumulator: DiagnosticAccumulator): self._diagnostic_accumulator = diagnostic_accumulator @@ -156,7 +157,7 @@ async def sync(self, ss: SelectorStore) -> AsyncGenerator[Update, None]: change_set_builder = ChangeSetBuilder() self._running = True - self._connection_attempt_started_monotonic = monotonic_seconds() + self._connection_attempt_start_time = monotonic_seconds() envid = None # A Start action with the X-LD-FD-Fallback: true header sets fallback_requested. @@ -173,7 +174,6 @@ async def sync(self, ss: SelectorStore) -> AsyncGenerator[Update, None]: # emit a fault with no-error. We can ignore this since we want # the connection to continue. if action.error is None: - self._connection_attempt_started_monotonic = monotonic_seconds() continue if action.headers is not None: @@ -190,12 +190,10 @@ async def sync(self, ss: SelectorStore) -> AsyncGenerator[Update, None]: break continue - if isinstance(action, Start): - self._connection_attempt_started_monotonic = monotonic_seconds() - if action.headers is not None: - envid = action.headers.get(_LD_ENVID_HEADER, envid) - if action.headers.get(_LD_FD_FALLBACK_HEADER) == 'true': - fallback_requested = True + if isinstance(action, Start) and action.headers is not None: + envid = action.headers.get(_LD_ENVID_HEADER, envid) + if action.headers.get(_LD_FD_FALLBACK_HEADER) == 'true': + fallback_requested = True if not isinstance(action, Event): continue @@ -204,7 +202,7 @@ async def sync(self, ss: SelectorStore) -> AsyncGenerator[Update, None]: update = process_message(action, change_set_builder, envid) if update is not None: self._record_stream_init(False) - self._connection_attempt_started_monotonic = None + self._connection_attempt_start_time = None if fallback_requested: # The completed update is the natural moment to honor # the latched directive: yield once with the signal, @@ -228,12 +226,11 @@ async def sync(self, ss: SelectorStore) -> AsyncGenerator[Update, None]: "Error while handling stream event; will restart stream: %s", e ) await self._sse.interrupt() - self._connection_attempt_started_monotonic = monotonic_seconds() yield with_fallback_signal(Update( state=DataSourceState.INTERRUPTED, error=DataSourceErrorInfo( - DataSourceErrorKind.UNKNOWN, 0, time.time(), str(e) + DataSourceErrorKind.UNKNOWN, 0, time(), str(e) ), fallback_to_fdv1=False, environment_id=envid, @@ -263,12 +260,12 @@ 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_started_monotonic is not None: - # The reported timestamp is a wall-clock value; the duration comes from - # the monotonic clock so a wall-clock step cannot distort it. + if self._diagnostic_accumulator and self._connection_attempt_start_time is not None: + # The reported timestamp stays on the wall clock; the duration is + # measured on the monotonic clock so a wall-clock step cannot distort it. current_time = current_time_millis() - elapsed = int((monotonic_seconds() - self._connection_attempt_started_monotonic) * 1000) - self._diagnostic_accumulator.record_stream_init(current_time, elapsed, failed) + 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]: """ @@ -285,9 +282,11 @@ async def _handle_error(self, error: Exception, envid: Optional[str]) -> Tuple[O if not self._running: return (None, False) # don't retry if we've been deliberately stopped - decision = classify_stream_error(error, envid) + decision = classify_stream_error( + error, self._sse.next_retry_delay, envid # type: ignore + ) self._record_stream_init(True) - self._connection_attempt_started_monotonic = monotonic_seconds() + self._connection_attempt_start_time = decision.next_attempt_start if decision.should_stop: await self.stop() diff --git a/ldclient/impl/datasourcev2/streaming.py b/ldclient/impl/datasourcev2/streaming.py index 3c5dd36f..ad1ad2d9 100644 --- a/ldclient/impl/datasourcev2/streaming.py +++ b/ldclient/impl/datasourcev2/streaming.py @@ -4,7 +4,7 @@ """ import json -import time +from time import time from typing import Callable, Generator, Optional, Tuple from urllib import parse @@ -132,7 +132,8 @@ def __init__(self, self._sse: Optional[SSEClient] = None self._running = False self._diagnostic_accumulator: Optional[DiagnosticAccumulator] = None - self._connection_attempt_started_monotonic: Optional[float] = None + # Monotonic-clock baseline for the current stream-init duration. + self._connection_attempt_start_time: Optional[float] = None def set_diagnostic_accumulator(self, diagnostic_accumulator: DiagnosticAccumulator): self._diagnostic_accumulator = diagnostic_accumulator @@ -164,7 +165,7 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, None]: change_set_builder = ChangeSetBuilder() self._running = True - self._connection_attempt_started_monotonic = monotonic_seconds() + self._connection_attempt_start_time = monotonic_seconds() envid = None # fallback_requested is set when a Start action carries @@ -182,7 +183,6 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, None]: # emit a fault with no-error. We can ignore this since we want # the connection to continue. if action.error is None: - self._connection_attempt_started_monotonic = monotonic_seconds() continue if action.headers is not None: @@ -199,12 +199,10 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, None]: break continue - if isinstance(action, Start): - self._connection_attempt_started_monotonic = monotonic_seconds() - if action.headers is not None: - envid = action.headers.get(_LD_ENVID_HEADER, envid) - if action.headers.get(_LD_FD_FALLBACK_HEADER) == 'true': - fallback_requested = True + if isinstance(action, Start) and action.headers is not None: + envid = action.headers.get(_LD_ENVID_HEADER, envid) + if action.headers.get(_LD_FD_FALLBACK_HEADER) == 'true': + fallback_requested = True if not isinstance(action, Event): continue @@ -213,7 +211,7 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, None]: update = process_message(action, change_set_builder, envid) if update is not None: self._record_stream_init(False) - self._connection_attempt_started_monotonic = None + self._connection_attempt_start_time = None if fallback_requested: # The completed update is the natural moment to honor # the latched directive: yield once with the signal, @@ -237,12 +235,11 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, None]: "Error while handling stream event; will restart stream: %s", e ) self._sse.interrupt() - self._connection_attempt_started_monotonic = monotonic_seconds() yield with_fallback_signal(Update( state=DataSourceState.INTERRUPTED, error=DataSourceErrorInfo( - DataSourceErrorKind.UNKNOWN, 0, time.time(), str(e) + DataSourceErrorKind.UNKNOWN, 0, time(), str(e) ), fallback_to_fdv1=False, environment_id=envid, @@ -262,10 +259,12 @@ def stop(self): self._sse.close() def _record_stream_init(self, failed: bool): - if self._diagnostic_accumulator and self._connection_attempt_started_monotonic is not None: + if self._diagnostic_accumulator and self._connection_attempt_start_time is not None: + # The reported timestamp stays on the wall clock; the duration is + # measured on the monotonic clock so a wall-clock step cannot distort it. current_time = current_time_millis() - elapsed = int((monotonic_seconds() - self._connection_attempt_started_monotonic) * 1000) - self._diagnostic_accumulator.record_stream_init(current_time, elapsed, failed) + 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]: """ @@ -282,9 +281,11 @@ def _handle_error(self, error: Exception, envid: Optional[str]) -> Tuple[Optiona if not self._running: return (None, False) # don't retry if we've been deliberately stopped - decision = classify_stream_error(error, envid) + decision = classify_stream_error( + error, self._sse.next_retry_delay, envid # type: ignore + ) self._record_stream_init(True) - self._connection_attempt_started_monotonic = monotonic_seconds() + self._connection_attempt_start_time = decision.next_attempt_start if decision.should_stop: self.stop() diff --git a/ldclient/impl/datasourcev2/streaming_common.py b/ldclient/impl/datasourcev2/streaming_common.py index f6aca4fd..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, @@ -132,23 +133,26 @@ class StreamErrorDecision: The values are computed without any I/O so both the sync and async synchronizers share one decision, then apply the side effects themselves: - they record a failed stream init, stop the synchronizer when ``should_stop`` - is set, and retry the connection when ``should_continue`` is set. + they record a failed stream init, set the connection attempt start time to + ``next_attempt_start``, stop the synchronizer when ``should_stop`` is set, + and retry the connection when ``should_continue`` is set. """ update: Update should_continue: bool should_stop: bool + next_attempt_start: Optional[float] def classify_stream_error( - error: Exception, envid: Optional[str] + error: Exception, next_retry_delay: float, envid: Optional[str] ) -> StreamErrorDecision: """ Decide how to react to a streaming connection error. The caller must first check that the synchronizer is still running; this - function assumes it is. It returns a decision the caller applies, recording - a failed stream init before doing so. + function assumes it is. It returns a decision the caller applies. The caller + always records a failed stream init before it sets the connection attempt + start time to the returned ``next_attempt_start``. """ if isinstance(error, json.decoder.JSONDecodeError): log.error("Unexpected error on stream connection: %s, will retry", error) @@ -164,9 +168,12 @@ def classify_stream_error( update=update, should_continue=True, should_stop=False, + next_attempt_start=monotonic_seconds() + next_retry_delay, ) if isinstance(error, HTTPStatusError): + next_attempt_start: Optional[float] = monotonic_seconds() + next_retry_delay + error_info = DataSourceErrorInfo( DataSourceErrorKind.ERROR_RESPONSE, error.status, @@ -188,6 +195,7 @@ def classify_stream_error( update=update, should_continue=False, should_stop=True, + next_attempt_start=next_attempt_start, ) http_error_message_result = http_error_message( @@ -211,6 +219,7 @@ def classify_stream_error( update=update, should_continue=False, should_stop=True, + next_attempt_start=None, ) log.warning(http_error_message_result) @@ -218,6 +227,7 @@ def classify_stream_error( update=update, should_continue=True, should_stop=False, + next_attempt_start=next_attempt_start, ) log.warning("Unexpected error on stream connection: %s, will retry", error) @@ -235,4 +245,5 @@ def classify_stream_error( update=update, should_continue=True, should_stop=False, + next_attempt_start=monotonic_seconds() + next_retry_delay, ) diff --git a/ldclient/impl/events/diagnostics.py b/ldclient/impl/events/diagnostics.py index e65c127e..9a799c75 100644 --- a/ldclient/impl/events/diagnostics.py +++ b/ldclient/impl/events/diagnostics.py @@ -16,7 +16,6 @@ def __init__(self, diagnostic_id): self._stream_inits = [] def record_stream_init(self, timestamp, duration, failed): - duration = max(0, duration) with self._state_lock: self._stream_inits.append({'timestamp': timestamp, 'durationMillis': duration, 'failed': failed}) diff --git a/ldclient/testing/impl/datasourcev2/test_async_streaming.py b/ldclient/testing/impl/datasourcev2/test_async_streaming.py index c417975c..b3c55f02 100644 --- a/ldclient/testing/impl/datasourcev2/test_async_streaming.py +++ b/ldclient/testing/impl/datasourcev2/test_async_streaming.py @@ -1,7 +1,6 @@ # pylint: disable=missing-docstring, too-few-public-methods import json -from types import SimpleNamespace from typing import AsyncIterable, List, Optional from unittest import mock @@ -550,72 +549,3 @@ async def test_envid_preserved_across_headerless_reconnect_start(): assert len(updates) == 2 assert updates[0].environment_id == 'my-env' assert updates[1].environment_id == 'my-env' - - -@pytest.mark.asyncio -async def test_failed_init_recorded_for_every_connect_attempt(): - # DIAG one-entry-per-attempt, exercised through the async run loop: three - # consecutive connect failures produce three failed records. - recorded = [] - src = make_streaming_data_source() - src.set_diagnostic_accumulator( - SimpleNamespace(record_stream_init=lambda ts, dur, failed: recorded.append((dur, failed))) - ) - - await collect_updates(src, [ - Fault(HTTPStatusError(503)), - Fault(HTTPStatusError(503)), - Fault(HTTPStatusError(503)), - ]) - - assert len(recorded) == 3 - assert all(failed for _, failed in recorded) - assert all(dur >= 0 for dur, _ in recorded) - - -@pytest.mark.asyncio -async def test_healthy_connection_drop_is_not_recorded_as_failed_init(): - # After a successful init, a mid-stream drop must NOT be recorded as a - # failed init (async run loop). - recorded = [] - src = make_streaming_data_source() - src.set_diagnostic_accumulator( - SimpleNamespace(record_stream_init=lambda ts, dur, failed: recorded.append((dur, failed))) - ) - - await collect_updates(src, [ - Start(), server_intent_event(IntentCode.TRANSFER_NONE), Fault(HTTPStatusError(503)), - ]) - - assert len(recorded) == 1 - assert recorded[0][1] is False - - -@pytest.mark.asyncio -async def test_reconnect_failure_after_clean_close_is_counted_without_stale_duration(monkeypatch): - # Async mirror of the sync clean-close teeth test: baseline re-stamped at the - # clean close (t=500.0), failure recorded 0.3s later -> 300ms. A stale - # baseline (left at the Start, t=200.0) would report 300300ms. - reads = iter([100.0, 200.0, 500.0, 500.3]) - last = [500.3] - - def fake_monotonic(): - try: - last[0] = next(reads) - except StopIteration: - pass - return last[0] - - monkeypatch.setattr(async_streaming, "monotonic_seconds", fake_monotonic) - - recorded = [] - src = make_streaming_data_source() - src.set_diagnostic_accumulator( - SimpleNamespace(record_stream_init=lambda ts, dur, failed: recorded.append((dur, failed))) - ) - - await collect_updates(src, [ - Start(), server_intent_event(IntentCode.TRANSFER_FULL), Fault(None), Fault(HTTPStatusError(503)), - ]) - - assert recorded == [(300, True)] diff --git a/ldclient/testing/impl/datasourcev2/test_stream_init_timing.py b/ldclient/testing/impl/datasourcev2/test_stream_init_timing.py index c7ae27ac..6cdfc426 100644 --- a/ldclient/testing/impl/datasourcev2/test_stream_init_timing.py +++ b/ldclient/testing/impl/datasourcev2/test_stream_init_timing.py @@ -1,116 +1,82 @@ """ -Tests for the monotonic stream-init timing in the FDv2 streaming synchronizers. +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. """ -import asyncio -import time +from unittest.mock import Mock -from ldclient.impl.datasourcev2 import \ - async_streaming as async_streaming_module -from ldclient.impl.datasourcev2 import streaming as streaming_module +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 -from ldclient.impl.events.diagnostics import _DiagnosticAccumulator - - -class _Recorder: - def __init__(self): - self.recorded = [] - - def record_stream_init(self, timestamp, duration, failed): - self.recorded.append((timestamp, duration, failed)) - - -class _FakeMonotonic: - def __init__(self, now: float = 1000.0): - self.now = now - - def __call__(self) -> float: - return self.now - - -def _bare(cls): - proc = object.__new__(cls) - recorder = _Recorder() - proc._diagnostic_accumulator = recorder - return proc, recorder +_CASES = [ + pytest.param(streaming, StreamingDataSource, id="sync"), + pytest.param(async_streaming, AsyncStreamingDataSource, id="async"), +] -def test_duration_unaffected_by_forward_wall_step(monkeypatch): - # A forward wall step used to inflate the reported duration by the step size. - for cls in (StreamingDataSource, AsyncStreamingDataSource): - proc, recorder = _bare(cls) - proc._connection_attempt_started_monotonic = time.monotonic() - real_time = time.time() - monkeypatch.setattr(time, "time", lambda: real_time + 3600) - proc._record_stream_init(False) - monkeypatch.undo() +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 - _, duration, failed = recorder.recorded[0] - assert failed is False - assert duration >= 0 - assert duration < 1000 +@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"]) -def test_backward_wall_step_does_not_zero_the_measurement(monkeypatch): - # A backward wall step used to drive the elapsed value negative, and the old - # clamp silently reported 0 instead of the real measurement. - for cls, module in ( - (StreamingDataSource, streaming_module), - (AsyncStreamingDataSource, async_streaming_module), - ): - proc, recorder = _bare(cls) - clock = _FakeMonotonic(1000.0) - monkeypatch.setattr(module, "monotonic_seconds", clock) - proc._connection_attempt_started_monotonic = clock() - clock.now = 1000.5 + source = _make_source(cls, monotonic["t"]) - real_time = time.time() - monkeypatch.setattr(time, "time", lambda: real_time - 120) - proc._record_stream_init(True) - monkeypatch.undo() + # 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 - _, duration, failed = recorder.recorded[0] - assert failed is True - assert duration == 500 + source._record_stream_init(False) + source._diagnostic_accumulator.record_stream_init.assert_called_once_with( + wall_millis["t"], 2000, False + ) -def test_error_handling_restamps_the_baseline(): - # _handle_error records the failed init, then re-stamps the baseline to now - # so the next (library-owned) reconnect attempt is still counted. It is not - # a projection and not cleared to None. - proc = object.__new__(StreamingDataSource) - proc._running = True - proc._diagnostic_accumulator = None - proc._connection_attempt_started_monotonic = 1234.0 - update, should_continue = proc._handle_error(Exception("boom"), None) +@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"]) - assert should_continue is True - assert update is not None - assert proc._connection_attempt_started_monotonic is not None - assert proc._connection_attempt_started_monotonic != 1234.0 + # 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) -def test_async_error_handling_restamps_the_baseline(): - proc = object.__new__(AsyncStreamingDataSource) - proc._running = True - proc._diagnostic_accumulator = None - proc._connection_attempt_started_monotonic = 1234.0 + source._diagnostic_accumulator.record_stream_init.assert_called_once_with( + wall_millis["t"], 0, True + ) - update, should_continue = asyncio.run(proc._handle_error(Exception("boom"), None)) - assert should_continue is True - assert update is not None - assert proc._connection_attempt_started_monotonic is not None - assert proc._connection_attempt_started_monotonic != 1234.0 +@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) -def test_diagnostic_accumulator_clamps_negative_durations(): - # The diagnostic schema requires durationMillis >= 0; the accumulator - # enforces it once for every call site. - accumulator = _DiagnosticAccumulator("diag-id") - accumulator.record_stream_init(123, -5, False) + source._record_stream_init(False) - assert accumulator._stream_inits[0]['durationMillis'] == 0 + source._diagnostic_accumulator.record_stream_init.assert_not_called() diff --git a/ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py b/ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py index eeeef6b2..216013b4 100644 --- a/ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py +++ b/ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py @@ -3,7 +3,6 @@ import json from abc import abstractmethod -from types import SimpleNamespace from typing import Iterable, List, Optional import pytest @@ -806,144 +805,3 @@ def test_envid_missing_when_no_headers(): assert len(updates) == 1 assert updates[0].state == DataSourceState.VALID assert updates[0].environment_id is None - - -def test_stream_init_duration_measured_from_start_not_run_start(monkeypatch): - # Regression for the projected-baseline bug (finding #41): the stream-init - # duration must be measured from the real connection moment (the Start - # action), not from run() start and not from a projection in _handle_error. - from ldclient.impl.datasourcev2 import streaming as streaming_module - - server_intent = ServerIntent( - payload=Payload( - id="id", - target=300, - code=IntentCode.TRANSFER_NONE, - reason="up-to-date", - ) - ) - intent_event = Event( - event=EventName.SERVER_INTENT, - data=json.dumps(server_intent.to_dict()), - ) - - # monotonic_seconds is read at: run() start, each Start action, and - # _record_stream_init. Script the three reads so the Start stamp (105.0) is - # distinct from run-start (100.0). The recorded duration must come out as - # 105.5 - 105.0 = 0.5s; a run-start baseline would give 5.5s. - reads = iter([100.0, 105.0, 105.5]) - last = [105.5] - - def fake_monotonic(): - try: - last[0] = next(reads) - except StopIteration: - pass - return last[0] - - monkeypatch.setattr(streaming_module, "monotonic_seconds", fake_monotonic) - - recorded = [] - synchronizer = make_streaming_data_source() - synchronizer._sse_client_builder = list_sse_client([Start(), intent_event]) - synchronizer.set_diagnostic_accumulator( - SimpleNamespace(record_stream_init=lambda ts, dur, failed: recorded.append((dur, failed))) - ) - - updates = list(synchronizer.sync(MockSelectorStore(Selector.no_selector()))) - - assert len(updates) == 1 - assert recorded == [(500, False)] - - -def _transfer_none_intent() -> Event: - return Event( - event=EventName.SERVER_INTENT, - data=json.dumps(ServerIntent(payload=Payload( - id="id", target=300, code=IntentCode.TRANSFER_NONE, reason="up-to-date", - )).to_dict()), - ) - - -def test_failed_init_recorded_for_every_connect_attempt(): - # DIAG requires one stream-init entry per connection attempt. Three - # consecutive connect failures must produce three failed records -- the - # baseline is re-stamped after each failure so the next attempt is counted. - recorded = [] - synchronizer = make_streaming_data_source() - synchronizer._sse_client_builder = list_sse_client([ - Fault(HTTPStatusError(503)), - Fault(HTTPStatusError(503)), - Fault(HTTPStatusError(503)), - ]) - synchronizer.set_diagnostic_accumulator( - SimpleNamespace(record_stream_init=lambda ts, dur, failed: recorded.append((dur, failed))) - ) - - list(synchronizer.sync(MockSelectorStore(Selector.no_selector()))) - - assert len(recorded) == 3 - assert all(failed for _, failed in recorded) - assert all(dur >= 0 for dur, _ in recorded) - - -def test_healthy_connection_drop_is_not_recorded_as_failed_init(): - # After a successful init, a mid-stream drop must NOT be recorded as a - # failed init -- the init already succeeded, so only the success is logged. - recorded = [] - synchronizer = make_streaming_data_source() - synchronizer._sse_client_builder = list_sse_client([ - Start(), _transfer_none_intent(), Fault(HTTPStatusError(503)), - ]) - synchronizer.set_diagnostic_accumulator( - SimpleNamespace(record_stream_init=lambda ts, dur, failed: recorded.append((dur, failed))) - ) - - list(synchronizer.sync(MockSelectorStore(Selector.no_selector()))) - - assert len(recorded) == 1 - assert recorded[0][1] is False - - -def test_reconnect_failure_after_clean_close_is_counted_without_stale_duration(monkeypatch): - # A connection that gets a Start but no completing payload, then is cleanly - # closed, re-stamps rather than leaving a stale baseline. A later failed - # reconnect is still counted, and its duration is measured from the clean - # close -- never a stale value spanning the closed connection and backoff. - # - # The monotonic clock is scripted so the Start, the clean close, and the - # reconnect fault land at known times: baseline is re-stamped at the clean - # close (t=500.0) and the failure is recorded 0.3s later, so the duration is - # exactly 300ms. A stale baseline (left at the Start, t=200.0) would instead - # report 300300ms -- the exact-value assert is what gives this test teeth. - from ldclient.impl.datasourcev2 import streaming as streaming_module - - reads = iter([100.0, 200.0, 500.0, 500.3]) - last = [500.3] - - def fake_monotonic(): - try: - last[0] = next(reads) - except StopIteration: - pass - return last[0] - - monkeypatch.setattr(streaming_module, "monotonic_seconds", fake_monotonic) - - server_intent = ServerIntent( - payload=Payload(id="id", target=300, code=IntentCode.TRANSFER_FULL, reason="cant-catchup"), - ) - intent_event = Event(event=EventName.SERVER_INTENT, data=json.dumps(server_intent.to_dict())) - - recorded = [] - synchronizer = make_streaming_data_source() - synchronizer._sse_client_builder = list_sse_client([ - Start(), intent_event, Fault(None), Fault(HTTPStatusError(503)), - ]) - synchronizer.set_diagnostic_accumulator( - SimpleNamespace(record_stream_init=lambda ts, dur, failed: recorded.append((dur, failed))) - ) - - list(synchronizer.sync(MockSelectorStore(Selector.no_selector()))) - - assert recorded == [(300, True)] From 0b804ffbb3ba16bdbae8d3314c076c0581c1955e Mon Sep 17 00:00:00 2001 From: Todd Anderson Date: Fri, 23 Oct 2026 14:33:20 -0400 Subject: [PATCH 5/5] refactor: Drop inline comments from the stream-init clock swap --- ldclient/impl/datasourcev2/async_streaming.py | 3 --- ldclient/impl/datasourcev2/streaming.py | 3 --- 2 files changed, 6 deletions(-) diff --git a/ldclient/impl/datasourcev2/async_streaming.py b/ldclient/impl/datasourcev2/async_streaming.py index f5701c6f..0f8428ab 100644 --- a/ldclient/impl/datasourcev2/async_streaming.py +++ b/ldclient/impl/datasourcev2/async_streaming.py @@ -123,7 +123,6 @@ def __init__( self._owned_session: Optional[aiohttp.ClientSession] = None self._running = False self._diagnostic_accumulator: Optional[DiagnosticAccumulator] = None - # Monotonic-clock baseline for the current stream-init duration. self._connection_attempt_start_time: Optional[float] = None def set_diagnostic_accumulator(self, diagnostic_accumulator: DiagnosticAccumulator): @@ -261,8 +260,6 @@ async def _close_owned_session(self): def _record_stream_init(self, failed: bool): if self._diagnostic_accumulator and self._connection_attempt_start_time is not None: - # The reported timestamp stays on the wall clock; the duration is - # measured on the monotonic clock so a wall-clock step cannot distort it. 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) diff --git a/ldclient/impl/datasourcev2/streaming.py b/ldclient/impl/datasourcev2/streaming.py index ad1ad2d9..5e9ab99a 100644 --- a/ldclient/impl/datasourcev2/streaming.py +++ b/ldclient/impl/datasourcev2/streaming.py @@ -132,7 +132,6 @@ def __init__(self, self._sse: Optional[SSEClient] = None self._running = False self._diagnostic_accumulator: Optional[DiagnosticAccumulator] = None - # Monotonic-clock baseline for the current stream-init duration. self._connection_attempt_start_time: Optional[float] = None def set_diagnostic_accumulator(self, diagnostic_accumulator: DiagnosticAccumulator): @@ -260,8 +259,6 @@ def stop(self): def _record_stream_init(self, failed: bool): if self._diagnostic_accumulator and self._connection_attempt_start_time is not None: - # The reported timestamp stays on the wall clock; the duration is - # measured on the monotonic clock so a wall-clock step cannot distort it. 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)