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