diff --git a/contract-tests/async_service.py b/contract-tests/async_service.py index b3c04d8a..65e9be5c 100644 --- a/contract-tests/async_service.py +++ b/contract-tests/async_service.py @@ -69,6 +69,8 @@ async def handle_status(request: aiohttp.web.Request) -> aiohttp.web.Response: 'migrations', 'persistent-data-store-redis', 'fdv1-fallback', + 'retry-conformance-fdv1-streaming', + 'retry-conformance-fdv1-polling', ] } return aiohttp.web.Response( diff --git a/contract-tests/service.py b/contract-tests/service.py index a8e93674..260fb77b 100644 --- a/contract-tests/service.py +++ b/contract-tests/service.py @@ -86,6 +86,8 @@ def status(): 'flag-change-listeners', 'flag-value-change-listeners', 'fdv1-fallback', + 'retry-conformance-fdv1-streaming', + 'retry-conformance-fdv1-polling', ] } return json.dumps(body), 200, {'Content-type': 'application/json'} diff --git a/ldclient/async_client.py b/ldclient/async_client.py index b741aa17..9e740b75 100644 --- a/ldclient/async_client.py +++ b/ldclient/async_client.py @@ -389,8 +389,8 @@ async def is_initialized(self) -> bool: If this returns false, it means that the client has not yet successfully connected to LaunchDarkly. It might still be in the process of starting up, or it might be attempting to reconnect after an - unsuccessful attempt, or it might have received an unrecoverable error (such as an invalid SDK key) - and given up. + unsuccessful attempt, or it might have received an error that needs to be fixed (such + as an invalid SDK key). This is a coroutine because determining readiness may query a persistent store. """ diff --git a/ldclient/async_config.py b/ldclient/async_config.py index b46b362e..0fc97dc1 100644 --- a/ldclient/async_config.py +++ b/ldclient/async_config.py @@ -15,6 +15,8 @@ from ldclient.config import ( DEFAULT_BASE_URI, DEFAULT_EVENTS_URI, + DEFAULT_INITIAL_RECONNECT_DELAY, + DEFAULT_POLL_INTERVAL, DEFAULT_STREAM_URI, GET_LATEST_FEATURES_PATH, STREAM_FLAGS_PATH, @@ -149,11 +151,11 @@ def __init__( flush_interval: float = 5, stream_uri: str = DEFAULT_STREAM_URI, stream: bool = True, - initial_reconnect_delay: float = 1, + initial_reconnect_delay: float = DEFAULT_INITIAL_RECONNECT_DELAY, defaults: dict = {}, send_events: Optional[bool] = None, update_processor_class: Optional[Callable[['AsyncConfig', AsyncFeatureStore, AsyncEvent], AsyncUpdateProcessor]] = None, - poll_interval: float = 30, + poll_interval: float = DEFAULT_POLL_INTERVAL, use_ldd: bool = False, feature_store: Optional[AsyncFeatureStore] = None, feature_requester_class=None, @@ -256,7 +258,7 @@ def __init__( self.__update_processor_class = update_processor_class self.__stream = stream self.__initial_reconnect_delay = initial_reconnect_delay - self.__poll_interval = max(poll_interval, 30.0) + self.__poll_interval = max(poll_interval, DEFAULT_POLL_INTERVAL) self.__use_ldd = use_ldd self.__feature_store = AsyncInMemoryFeatureStore() if not feature_store else feature_store self.__event_processor_class = event_processor_class diff --git a/ldclient/client.py b/ldclient/client.py index 8433fc1d..869cabd6 100644 --- a/ldclient/client.py +++ b/ldclient/client.py @@ -317,10 +317,11 @@ def is_initialized(self) -> bool: If this returns false, it means the client has not yet obtained any flag data. It might still be starting up, or attempting to reconnect after an unsuccessful attempt, or it might have received - an unrecoverable error (such as an invalid SDK key) and given up. In this state, feature flag - evaluations will return default values -- unless you are using a persistent store integration and - flag data had already been stored by a successfully connected SDK in the past. You can use - :attr:`data_source_status_provider` to get information on errors, or to wait for a successful retry. + an error that needs to be fixed (such as an invalid SDK key). In this state, feature flag + evaluations will return default values -- unless you are using a persistent store integration + and flag data had already been stored by a successfully connected SDK in the past. You can use + :attr:`data_source_status_provider` to get information on errors, or to wait for a + successful retry. :return: true if the client is initialized and has flag data available """ diff --git a/ldclient/config.py b/ldclient/config.py index d7ac76a2..2e9c8e96 100644 --- a/ldclient/config.py +++ b/ldclient/config.py @@ -36,6 +36,11 @@ DEFAULT_EVENTS_URI = 'https://events.launchdarkly.com' DEFAULT_STREAM_URI = 'https://stream.launchdarkly.com' +# Defaults, in seconds, for the two configurable data source intervals. The +# poll interval is also its own minimum. +DEFAULT_INITIAL_RECONNECT_DELAY = 1 +DEFAULT_POLL_INTERVAL = 30 + class BigSegmentsConfig: """Configuration options related to Big Segments. @@ -295,11 +300,11 @@ def __init__( flush_interval: float = 5, stream_uri: str = DEFAULT_STREAM_URI, stream: bool = True, - initial_reconnect_delay: float = 1, + initial_reconnect_delay: float = DEFAULT_INITIAL_RECONNECT_DELAY, defaults: dict = {}, send_events: Optional[bool] = None, update_processor_class: Optional[Callable[['Config', FeatureStore, Event], UpdateProcessor]] = None, - poll_interval: float = 30, + poll_interval: float = DEFAULT_POLL_INTERVAL, use_ldd: bool = False, feature_store: Optional[FeatureStore] = None, feature_requester_class=None, @@ -402,7 +407,7 @@ def __init__( self.__update_processor_class = update_processor_class self.__stream = stream self.__initial_reconnect_delay = initial_reconnect_delay - self.__poll_interval = max(poll_interval, 30.0) + self.__poll_interval = max(poll_interval, DEFAULT_POLL_INTERVAL) self.__use_ldd = use_ldd self.__feature_store = InMemoryFeatureStore() if not feature_store else feature_store self.__event_processor_class = event_processor_class diff --git a/ldclient/impl/aio/transport.py b/ldclient/impl/aio/transport.py index e9aa374f..947d13d3 100644 --- a/ldclient/impl/aio/transport.py +++ b/ldclient/impl/aio/transport.py @@ -118,11 +118,16 @@ def __init__(self, config, session: Optional[aiohttp.ClientSession] = None, prox self._http_options = http_options if http_options is not None else config.http self._proxy = proxy if proxy is not None else (self._http_options.http_proxy or None) - def create(self, url: str, initial_retry_delay: float, query_params=None) -> AsyncSSEClient: - """Builds an SSE client for the given stream URL. Headers, timeouts, - proxy settings, and the retry/backoff policy come from the SDK config. - ``query_params`` is an optional zero-argument callable evaluated on - each (re)connect to produce additional query string parameters.""" + def create(self, url: str, initial_retry_delay: float, query_params=None, sdk_managed_retry: bool = False) -> AsyncSSEClient: + """Builds an SSE client for the given stream URL. Headers, timeouts and + proxy settings come from the SDK config. ``query_params`` is an + optional zero-argument callable evaluated on each (re)connect to + produce additional query string parameters. + + ``sdk_managed_retry`` moves the delay between connection attempts to + the caller. The SSE client then never waits, and + ``initial_retry_delay`` is ignored. When it is false, the SSE client + backs off on its own.""" base_headers = _base_headers(self._config, ASYNC_USER_AGENT) aiohttp_request_options: dict = { "timeout": aiohttp.ClientTimeout( @@ -134,6 +139,24 @@ def create(self, url: str, initial_retry_delay: float, query_params=None) -> Asy proxy = self._proxy or _get_proxy_url(url) if proxy: aiohttp_request_options["proxy"] = proxy + if sdk_managed_retry: + # The SSE client's retry is disabled; the SDK owns the delay. The base + # strategy must be passed: omitting it selects the library's backoff. + retry_options: dict = { + "initial_retry_delay": 0, + "retry_delay_strategy": RetryDelayStrategy(), + "retry_delay_reset_threshold": 0, + } + else: + retry_options = { + "initial_retry_delay": initial_retry_delay, + "retry_delay_strategy": RetryDelayStrategy.default( + max_delay=MAX_RETRY_DELAY, + backoff_multiplier=2, + jitter_multiplier=JITTER_RATIO, + ), + "retry_delay_reset_threshold": BACKOFF_RESET_INTERVAL, + } return AsyncSSEClient( connect=AsyncConnectStrategy.http( url=url, @@ -143,12 +166,6 @@ def create(self, url: str, initial_retry_delay: float, query_params=None) -> Asy query_params=query_params, ), error_strategy=ErrorStrategy.always_continue(), # we'll make error-handling decisions when we see a Fault - initial_retry_delay=initial_retry_delay, - retry_delay_strategy=RetryDelayStrategy.default( - max_delay=MAX_RETRY_DELAY, - backoff_multiplier=2, - jitter_multiplier=JITTER_RATIO, - ), - retry_delay_reset_threshold=BACKOFF_RESET_INTERVAL, logger=log, + **retry_options, ) diff --git a/ldclient/impl/datasource/async_polling.py b/ldclient/impl/datasource/async_polling.py index 2c2e56e7..91f24915 100644 --- a/ldclient/impl/datasource/async_polling.py +++ b/ldclient/impl/datasource/async_polling.py @@ -10,10 +10,15 @@ from ldclient.async_config import AsyncConfig from ldclient.impl.aio.concurrency import AsyncEvent, AsyncRepeatingTask from ldclient.impl.datasource.datasource_common import sink_or_store +from ldclient.impl.retry import ( + FailureKind, + RetryState, + classify_http_status, + for_polling +) from ldclient.impl.util import ( UnsuccessfulResponseException, - http_error_message, - is_http_error_recoverable, + http_error_description, log ) from ldclient.interfaces import ( @@ -27,13 +32,22 @@ class AsyncPollingUpdateProcessor(AsyncUpdateProcessor): - def __init__(self, config: AsyncConfig, requester: AsyncFeatureRequester, store: AsyncFeatureStore, ready: AsyncEvent): + """Polls LaunchDarkly for flag data on its own background task. + + The loop reads its wait from the retry state, which ``_fetch_and_store`` + updates, so a failure can push the next poll further out than the poll + interval. See :mod:`ldclient.impl.retry`. + """ + + def __init__(self, config: AsyncConfig, requester: AsyncFeatureRequester, store: AsyncFeatureStore, ready: AsyncEvent, retry_state: Optional[RetryState] = None): self._config = config self._data_source_update_sink = config.data_source_update_sink self._requester = requester self._store = store self._ready = ready - self._task = AsyncRepeatingTask.at_interval("ldclient.datasource.polling", config.poll_interval, 0, self._fetch_and_store) + self._retry = retry_state or for_polling(config.poll_interval) + # No initial delay: the first poll is immediate. + self._task = AsyncRepeatingTask("ldclient.datasource.polling", self._retry, 0, self._fetch_and_store) def start(self): log.info("Starting AsyncPollingUpdateProcessor with request interval: " + str(self._config.poll_interval)) @@ -43,48 +57,53 @@ def initialized(self): return self._ready.is_set() and self._store.initialized async def stop(self): - self.__stop_with_error_info(None) - # Wait for the current poll to finish before closing the transport, so we do - # not close it while a request is still using it. The close is in a finally - # so an owned transport is still released if stop() is cancelled mid-wait. - try: - await self._task.wait_stopped() - finally: - await self._requester.close() - - def __stop_with_error_info(self, error: Optional[DataSourceErrorInfo]): log.info("Stopping AsyncPollingUpdateProcessor") self._task.stop() - if self._data_source_update_sink is None: - return + if self._data_source_update_sink is not None: + self._data_source_update_sink.update_status(DataSourceState.OFF, None) - self._data_source_update_sink.update_status(DataSourceState.OFF, error) + # Do not close the transport while an in-flight request still uses it. + try: + await self._task.wait_stopped() + finally: + await self._requester.close() - async def _fetch_and_store(self): + async def _fetch_and_store(self) -> None: + """Makes one poll request and records the outcome on the retry state.""" try: all_data = await self._requester.get_all_data() await sink_or_store(self._data_source_update_sink, self._store).init(all_data) + + if self._data_source_update_sink is not None: + self._data_source_update_sink.update_status(DataSourceState.VALID, None) + + # Report the status before signaling readiness, so a caller that + # wakes on readiness cannot still read INITIALIZING. if not self._ready.is_set() and self._store.initialized: log.info("AsyncPollingUpdateProcessor initialized ok") self._ready.set() - if self._data_source_update_sink is not None: - self._data_source_update_sink.update_status(DataSourceState.VALID, None) + self._retry.record_success() + return except UnsuccessfulResponseException as e: + kind = classify_http_status(e.status) error_info = DataSourceErrorInfo(DataSourceErrorKind.ERROR_RESPONSE, e.status, time.time(), str(e)) + description = "Received %s for polling request" % http_error_description(e.status) + level = log.error if kind is FailureKind.UNEXPECTED else log.warning + stacktrace = None + except Exception as e: + kind = FailureKind.NORMAL + error_info = DataSourceErrorInfo(DataSourceErrorKind.UNKNOWN, 0, time.time(), str(e)) + description = "Error encountered when updating flags: %s" % e + level = log.error + # The exception is passed explicitly: by the time the message is + # logged, the handler has exited and exc_info() is empty. + stacktrace = e - http_error_message_result = http_error_message(e.status, "polling request") - if not is_http_error_recoverable(e.status): - log.error(http_error_message_result) - self._ready.set() # if client is initializing, make it stop waiting; has no effect if already inited - self.__stop_with_error_info(error_info) - else: - log.warning(http_error_message_result) + self._retry.record_failure(kind) + delay = self._retry.next_delay + level("%s - will retry in %.1fs" % (description, delay), exc_info=stacktrace) - if self._data_source_update_sink is not None: - self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, error_info) - except Exception as e: - log.exception('Error: Exception encountered when updating flags. %s' % e) - if self._data_source_update_sink is not None: - self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, DataSourceErrorInfo(DataSourceErrorKind.UNKNOWN, 0, time.time(), str(e))) + if self._data_source_update_sink is not None: + self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, error_info) diff --git a/ldclient/impl/datasource/async_status.py b/ldclient/impl/datasource/async_status.py index 1bebc281..29bccdb0 100644 --- a/ldclient/impl/datasource/async_status.py +++ b/ldclient/impl/datasource/async_status.py @@ -72,6 +72,11 @@ def update_status(self, new_state: DataSourceState, new_error: Optional[DataSour old_status = self.__status + # OFF is terminal. A poll or stream connection that was still in + # flight when the data source stopped must not report after it. + if old_status.state == DataSourceState.OFF: + return + if new_state == DataSourceState.INTERRUPTED and old_status.state == DataSourceState.INITIALIZING: new_state = DataSourceState.INITIALIZING diff --git a/ldclient/impl/datasource/async_streaming.py b/ldclient/impl/datasource/async_streaming.py index 002d7b65..397697d5 100644 --- a/ldclient/impl/datasource/async_streaming.py +++ b/ldclient/impl/datasource/async_streaming.py @@ -4,9 +4,10 @@ # currently excluded from documentation - see docs/README.md +import asyncio import json import time -from typing import Any, Optional +from typing import Any, Callable, Optional from urllib import parse from ld_eventsource.actions import Event, Fault, Start @@ -16,14 +17,17 @@ from ldclient.impl.aio.transport import AsyncSSEFactory, make_client_session from ldclient.impl.datasource.datasource_common import ( STREAM_ALL_PATH, + StreamClosedError, parse_path, sink_or_store ) -from ldclient.impl.util import ( - http_error_message, - is_http_error_recoverable, - log +from ldclient.impl.retry import ( + FailureKind, + RetryState, + classify_http_status, + for_streaming ) +from ldclient.impl.util import http_error_description, log from ldclient.interfaces import ( AsyncUpdateProcessor, DataSourceErrorInfo, @@ -34,7 +38,13 @@ class AsyncStreamingUpdateProcessor(AsyncUpdateProcessor): - def __init__(self, config, store, ready, diagnostic_accumulator, sse_factory: Optional[AsyncSSEFactory] = None): + """Reads flag data from LaunchDarkly's streaming endpoint on a background task. + + The SDK owns the delay between connection attempts rather than the SSE + client; see :mod:`ldclient.impl.retry`. + """ + + def __init__(self, config, store, ready, diagnostic_accumulator, sse_factory: Optional[AsyncSSEFactory] = None, retry_state: Optional[RetryState] = None): self._uri = config.stream_base_uri + STREAM_ALL_PATH if config.payload_filter_key is not None: self._uri += '?%s' % parse.urlencode({'filter': config.payload_filter_key}) @@ -51,13 +61,16 @@ def __init__(self, config, store, ready, diagnostic_accumulator, sse_factory: Op self._sse_factory = sse_factory self._owned_session = None self._sse: Any = None - self._connection_attempt_start_time = None + self._connection_attempt_start_time: Optional[float] = None self._runner = AsyncTaskRunner() self._started = False + self._retry = retry_state or for_streaming(config.initial_reconnect_delay) + self._interrupted_by_sdk = False def start(self): if self._started: - raise RuntimeError("processors can only be started once") + log.info("AsyncStreamingUpdateProcessor has already been started; ignoring") + return self._started = True self._runner.spawn("ldclient.datasource.streaming", self._run) @@ -72,31 +85,39 @@ async def _run(self): log.info("Starting AsyncStreamingUpdateProcessor connecting to uri: " + self._uri) self._running = True try: - self._sse = self._sse_factory.create(self._uri, self._config.initial_reconnect_delay) + self._sse = self._sse_factory.create(self._uri, self._config.initial_reconnect_delay, sdk_managed_retry=True) self._connection_attempt_start_time = time.time() async for action in self._sse.all: if isinstance(action, Start): + # interrupt() is a no-op when the connection has already gone, so + # clear a stale flag here rather than swallow the next real close. + self._interrupted_by_sdk = False + # On reconnect after an error the timer was cleared; reset it here. # For the initial connect the pre-loop timestamp is already set. if self._connection_attempt_start_time is None: self._connection_attempt_start_time = time.time() elif isinstance(action, Event): message_ok = False + message_handled = False try: message_ok = await self._process_message(action) + message_handled = True except json.decoder.JSONDecodeError as e: log.info("Error while handling stream event; will restart stream: %s" % e) - await self._sse.interrupt() + await self._interrupt_stream() - await self._handle_error(e) + if not await self._handle_error(e): + break except Exception as e: log.warning("Error while handling stream event; will restart stream: %s" % e) - await self._sse.interrupt() + await self._interrupt_stream() - if self._data_source_update_sink is not None: - error_info = DataSourceErrorInfo(DataSourceErrorKind.UNKNOWN, 0, time.time(), str(e)) + if not await self._handle_error(e): + break - self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, error_info) + if message_handled: + self._retry.record_success() if message_ok: self._record_stream_init(False) @@ -109,9 +130,14 @@ async def _run(self): log.info("AsyncStreamingUpdateProcessor initialized ok.") self._ready.set() elif isinstance(action, Fault): - # If the SSE client detects the stream has closed, then it will emit a fault with no-error. We can - # ignore this since we want the connection to continue. + # A Fault with no error is a clean close. An interrupt the + # SDK asked for is not a failure. if action.error is None: + if self._interrupted_by_sdk: + self._interrupted_by_sdk = False + continue + if not await self._handle_error(StreamClosedError()): + break continue if not await self._handle_error(action.error): @@ -140,23 +166,29 @@ def _record_stream_init(self, failed: bool): self._diagnostic_accumulator.record_stream_init(current_time, elapsed if elapsed >= 0 else 0, failed) async def stop(self): - # Cancel the run task first: otherwise, if stop() is called before _run has executed, the - # loop could run _run at __stop_with_error_info's await and create a fresh SSE connection - # against the session we're closing. Once the runner is stopped, teardown is safe. - await self._runner.stop_all() - await self.__stop_with_error_info(None) - - async def __stop_with_error_info(self, error: Optional[DataSourceErrorInfo]): log.info("Stopping AsyncStreamingUpdateProcessor") self._running = False + + # OFF means an explicit shutdown. No stream failure produces it. It is + # reported before teardown, so a slow close cannot hold back the status + # that tells a waiter to give up. The sink drops anything after OFF. + if self._data_source_update_sink is not None: + self._data_source_update_sink.update_status(DataSourceState.OFF, None) + + # Cancel the run task before tearing down the rest, preventing _run from + # starting a fresh SSE connection against the session we are closing. + await self._runner.stop_all() + if self._sse: await self._sse.close() await self._close_owned_session() - if self._data_source_update_sink is None: - return - - self._data_source_update_sink.update_status(DataSourceState.OFF, error) + async def _interrupt_stream(self): + """Drops the stream connection so the next read reconnects. The SSE + client reports the close as a Fault with no error, and the flag tells + the loop that this one is ours and is already accounted for.""" + self._interrupted_by_sdk = True + await self._sse.interrupt() def initialized(self): return self._running and self._ready.is_set() is True and self._store.initialized is True @@ -198,46 +230,56 @@ async def _process_message(self, msg: Event) -> bool: # Returns true to continue, false to stop async def _handle_error(self, error: Exception) -> bool: + """Records a stream failure, reports it, and waits before the retry. + + Returns True once the wait is over, or False if the processor was + stopped. No failure ever ends the stream by itself. The wait is + interrupted by cancelling the task, which matters because the extended + regime can ask for an hour. + """ if not self._running: return False # don't retry if we've been deliberately stopped - if isinstance(error, json.decoder.JSONDecodeError): - error_info = DataSourceErrorInfo(DataSourceErrorKind.INVALID_DATA, 0, time.time(), str(error)) + self._record_stream_init(True) - log.error("Unexpected error on stream connection: %s, will retry" % error) - self._record_stream_init(True) - self._connection_attempt_start_time = None + level: Callable[..., None] - if self._data_source_update_sink is not None: - self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, error_info) + if isinstance(error, json.decoder.JSONDecodeError): + kind = FailureKind.NORMAL + error_info = DataSourceErrorInfo(DataSourceErrorKind.INVALID_DATA, 0, time.time(), str(error)) + description = "Unparseable data on stream connection: %s" % error + level = log.error elif isinstance(error, HTTPStatusError): - self._record_stream_init(True) - self._connection_attempt_start_time = None - + kind = classify_http_status(error.status) error_info = DataSourceErrorInfo(DataSourceErrorKind.ERROR_RESPONSE, error.status, time.time(), str(error)) + description = "Received %s for stream connection" % http_error_description(error.status) + level = log.error if kind is FailureKind.UNEXPECTED else log.warning + elif isinstance(error, StreamClosedError): + kind = FailureKind.NORMAL + error_info = DataSourceErrorInfo(DataSourceErrorKind.NETWORK_ERROR, 0, time.time(), str(error)) + description = "The server closed the stream connection" + level = log.warning + else: + kind = FailureKind.NORMAL + error_info = DataSourceErrorInfo(DataSourceErrorKind.UNKNOWN, 0, time.time(), str(error)) + # no stacktrace here because, for a typical connection error, it'll just be a lengthy tour of HTTP client internals + description = "Error on stream connection: %s" % error + level = log.warning - http_error_message_result = http_error_message(error.status, "stream connection") - if not is_http_error_recoverable(error.status): - log.error(http_error_message_result) - self._running = False - self._ready.set() # if client is initializing, make it stop waiting; has no effect if already inited - await self.__stop_with_error_info(error_info) - return False - else: - log.warning(http_error_message_result) + self._retry.record_failure(kind) + delay = self._retry.next_delay + level("%s - will retry in %.1fs" % (description, delay)) - if self._data_source_update_sink is not None: - self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, error_info) - else: - log.warning("Unexpected error on stream connection: %s, will retry" % error) - self._record_stream_init(True) - self._connection_attempt_start_time = None + if self._data_source_update_sink is not None: + self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, error_info) - if self._data_source_update_sink is not None: - self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, DataSourceErrorInfo(DataSourceErrorKind.UNKNOWN, 0, time.time(), str(error))) - # no stacktrace here because, for a typical connection error, it'll just be a lengthy tour of HTTP client internals - self._connection_attempt_start_time = time.time() + self._sse.next_retry_delay - return True + if delay > 0: + await asyncio.sleep(delay) + + # Read after the wait, so a clock change during it cannot skew the + # stream-init latency we report. + self._connection_attempt_start_time = time.time() + return self._running # magic methods for "with" statement (used in testing) async def __aenter__(self): diff --git a/ldclient/impl/datasource/datasource_common.py b/ldclient/impl/datasource/datasource_common.py index aa4da878..ad8de2a0 100644 --- a/ldclient/impl/datasource/datasource_common.py +++ b/ldclient/impl/datasource/datasource_common.py @@ -5,10 +5,16 @@ # currently excluded from documentation - see docs/README.md from collections import namedtuple -from typing import Mapping, Optional, Protocol, runtime_checkable +from typing import ( + Mapping, + Optional, + Protocol, + TypeVar, + Union, + runtime_checkable +) from ldclient.impl.util import _LD_ENVID_HEADER -from ldclient.interfaces import DataSourceUpdateSink, FeatureStore from ldclient.versioned_data_kind import FEATURES, SEGMENTS STREAM_ALL_PATH = '/all' @@ -17,7 +23,28 @@ ParsedPath = namedtuple('ParsedPath', ['kind', 'key']) -def sink_or_store(sink: Optional[DataSourceUpdateSink], store: FeatureStore): +class StreamClosedError(Exception): + """The stream connection closed cleanly, and the SDK did not ask for it. + + The service normally leaves the connection open, so a close the SDK did + not ask for is a connection failure. The SDK backs off before it + reconnects, rather than reconnecting at once. + + It is a NORMAL failure, not an UNEXPECTED one. A load balancer draining + during a rolling deploy closes streams cleanly, and putting that in the + extended regime would take a whole fleet out of service for up to an + hour. + """ + + def __init__(self): + super().__init__("the server closed the stream connection") + + +_Sink = TypeVar('_Sink') +_Store = TypeVar('_Store') + + +def sink_or_store(sink: Optional[_Sink], store: _Store) -> Union[_Sink, _Store]: """ The original implementation of the data sources relied on the feature store directly, which we are trying to move away from. Customers who might have diff --git a/ldclient/impl/datasource/polling.py b/ldclient/impl/datasource/polling.py index 8b33bc3d..2eacc6ad 100644 --- a/ldclient/impl/datasource/polling.py +++ b/ldclient/impl/datasource/polling.py @@ -14,10 +14,15 @@ sink_or_store ) from ldclient.impl.repeating_task import RepeatingTask +from ldclient.impl.retry import ( + FailureKind, + RetryState, + classify_http_status, + for_polling +) from ldclient.impl.util import ( UnsuccessfulResponseException, - http_error_message, - is_http_error_recoverable, + http_error_description, log ) from ldclient.interfaces import ( @@ -38,13 +43,21 @@ def get_all_data_with_headers(self) -> Tuple[Any, Optional[Mapping[str, str]]]: class PollingUpdateProcessor(UpdateProcessor): - def __init__(self, config: Config, requester: FeatureRequester, store: FeatureStore, ready: Event): + """Polls LaunchDarkly for flag data on its own worker thread. + + The task reads its wait from the retry state, which ``_poll`` updates, so a + failure can push the next poll further out than the poll interval. See + :mod:`ldclient.impl.retry`. + """ + + def __init__(self, config: Config, requester: FeatureRequester, store: FeatureStore, ready: Event, retry_state: Optional[RetryState] = None): self._config = config self._data_source_update_sink: Optional[DataSourceUpdateSink] = config.data_source_update_sink self._requester = requester self._store = store self._ready = ready - self._task = RepeatingTask.at_interval("ldclient.datasource.polling", config.poll_interval, 0, self._poll) + self._retry = retry_state or for_polling(config.poll_interval) + self._task = RepeatingTask("ldclient.datasource.polling", self._retry, 0, self._poll) def start(self): log.info("Starting PollingUpdateProcessor with request interval: " + str(self._config.poll_interval)) @@ -54,46 +67,53 @@ def initialized(self): return self._ready.is_set() is True and self._store.initialized is True def stop(self): - self.__stop_with_error_info(None) - - def __stop_with_error_info(self, error: Optional[DataSourceErrorInfo]): log.info("Stopping PollingUpdateProcessor") self._task.stop() if self._data_source_update_sink is None: return - self._data_source_update_sink.update_status(DataSourceState.OFF, error) + self._data_source_update_sink.update_status(DataSourceState.OFF, None) - def _poll(self): + def _poll(self) -> None: + """Makes one poll request and records the outcome on the retry state.""" try: (all_data, headers) = self._get_all_data_with_headers() record_environment_id(self._data_source_update_sink, headers) sink_or_store(self._data_source_update_sink, self._store).init(all_data) + + if self._data_source_update_sink is not None: + self._data_source_update_sink.update_status(DataSourceState.VALID, None) + + # Report the status before signaling readiness, so a caller that + # wakes on readiness cannot still read INITIALIZING. if not self._ready.is_set() and self._store.initialized: log.info("PollingUpdateProcessor initialized ok") self._ready.set() - if self._data_source_update_sink is not None: - self._data_source_update_sink.update_status(DataSourceState.VALID, None) + self._retry.record_success() + return except UnsuccessfulResponseException as e: + kind = classify_http_status(e.status) error_info = DataSourceErrorInfo(DataSourceErrorKind.ERROR_RESPONSE, e.status, time.time(), str(e)) - - http_error_message_result = http_error_message(e.status, "polling request") - if not is_http_error_recoverable(e.status): - log.error(http_error_message_result) - self._ready.set() # if client is initializing, make it stop waiting; has no effect if already inited - self.__stop_with_error_info(error_info) - else: - log.warning(http_error_message_result) - - if self._data_source_update_sink is not None: - self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, error_info) + description = "Received %s for polling request" % http_error_description(e.status) + level = log.error if kind is FailureKind.UNEXPECTED else log.warning + stacktrace = None except Exception as e: - log.exception('Error: Exception encountered when updating flags. %s' % e) - - if self._data_source_update_sink is not None: - self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, DataSourceErrorInfo(DataSourceErrorKind.UNKNOWN, 0, time.time(), str(e))) + kind = FailureKind.NORMAL + error_info = DataSourceErrorInfo(DataSourceErrorKind.UNKNOWN, 0, time.time(), str(e)) + description = "Error encountered when updating flags: %s" % e + level = log.error + # The exception is passed explicitly: by the time the message is + # logged, the handler has exited and exc_info() is empty. + stacktrace = e + + self._retry.record_failure(kind) + delay = self._retry.next_delay + level("%s - will retry in %.1fs" % (description, delay), exc_info=stacktrace) + + if self._data_source_update_sink is not None: + self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, error_info) def _get_all_data_with_headers(self) -> Tuple[Any, Optional[Mapping[str, str]]]: """ diff --git a/ldclient/impl/datasource/status.py b/ldclient/impl/datasource/status.py index c4d046d7..cec2ed24 100644 --- a/ldclient/impl/datasource/status.py +++ b/ldclient/impl/datasource/status.py @@ -80,6 +80,11 @@ def update_status(self, new_state: DataSourceState, new_error: Optional[DataSour with self.__lock.write(): old_status = self.__status + # OFF is terminal. A poll or stream connection that was still in + # flight when the data source stopped must not report after it. + if old_status.state == DataSourceState.OFF: + return + if new_state == DataSourceState.INTERRUPTED and old_status.state == DataSourceState.INITIALIZING: new_state = DataSourceState.INITIALIZING diff --git a/ldclient/impl/datasource/streaming.py b/ldclient/impl/datasource/streaming.py index e5496147..2e48e918 100644 --- a/ldclient/impl/datasource/streaming.py +++ b/ldclient/impl/datasource/streaming.py @@ -1,7 +1,9 @@ import json import time +from threading import TIMEOUT_MAX +from threading import Event as ThreadEvent from threading import Thread -from typing import Optional +from typing import Callable, Optional from urllib import parse from ld_eventsource import SSEClient @@ -15,16 +17,19 @@ from ldclient.impl.datasource.datasource_common import ( STREAM_ALL_PATH, + StreamClosedError, parse_path, record_environment_id, sink_or_store ) from ldclient.impl.http import HTTPFactory, _http_factory -from ldclient.impl.util import ( - http_error_message, - is_http_error_recoverable, - log +from ldclient.impl.retry import ( + FailureKind, + RetryState, + classify_http_status, + for_streaming ) +from ldclient.impl.util import http_error_description, log from ldclient.interfaces import ( DataSourceErrorInfo, DataSourceErrorKind, @@ -37,13 +42,15 @@ # stream will keep this from triggering stream_read_timeout = 5 * 60 -MAX_RETRY_DELAY = 30 -BACKOFF_RESET_INTERVAL = 60 -JITTER_RATIO = 0.5 - class StreamingUpdateProcessor(Thread, UpdateProcessor): - def __init__(self, config, store, ready, diagnostic_accumulator): + """Reads flag data from LaunchDarkly's streaming endpoint on its own thread. + + The SDK owns the delay between connection attempts rather than the SSE + client; see :meth:`_create_sse_client` and :mod:`ldclient.impl.retry`. + """ + + def __init__(self, config, store, ready, diagnostic_accumulator, retry_state: Optional[RetryState] = None): Thread.__init__(self, name="ldclient.datasource.streaming") self.daemon = True self._uri = config.stream_base_uri + STREAM_ALL_PATH @@ -55,53 +62,79 @@ def __init__(self, config, store, ready, diagnostic_accumulator): self._running = False self._ready = ready self._diagnostic_accumulator = diagnostic_accumulator - self._connection_attempt_start_time = None + self._connection_attempt_start_time: Optional[float] = None + self._retry = retry_state or for_streaming(config.initial_reconnect_delay) + self._sse: Optional[SSEClient] = None + self._stop_event = ThreadEvent() + self._interrupted_by_sdk = False def run(self): log.info("Starting StreamingUpdateProcessor connecting to uri: " + self._uri) self._running = True self._sse = self._create_sse_client() + + # stop() may have run before the client existed, in which case it had + # nothing to close. Never read a connection nobody is left to close. + if self._stop_event.is_set(): + self._sse.close() + return + self._connection_attempt_start_time = time.time() - for action in self._sse.all: - if isinstance(action, Start): - record_environment_id(self._data_source_update_sink, action.headers) - elif isinstance(action, Event): - message_ok = False - try: - message_ok = self._process_message(sink_or_store(self._data_source_update_sink, self._store), action) - except json.decoder.JSONDecodeError as e: - log.info("Error while handling stream event; will restart stream: %s" % e) - self._sse.interrupt() - - self._handle_error(e) - except Exception as e: - log.info("Error while handling stream event; will restart stream: %s" % e) - self._sse.interrupt() - - if self._data_source_update_sink is not None: - error_info = DataSourceErrorInfo(DataSourceErrorKind.UNKNOWN, 0, time.time(), str(e)) - - self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, error_info) - - if message_ok: - self._record_stream_init(False) - self._connection_attempt_start_time = None - - if self._data_source_update_sink is not None: - self._data_source_update_sink.update_status(DataSourceState.VALID, None) - - if not self._ready.is_set(): - log.info("StreamingUpdateProcessor initialized ok.") - self._ready.set() - elif isinstance(action, Fault): - # If the SSE client detects the stream has closed, then it will emit a fault with no-error. We can - # ignore this since we want the connection to continue. - if action.error is None: - continue - - if not self._handle_error(action.error): - break - self._sse.close() + try: + for action in self._sse.all: + if isinstance(action, Start): + # interrupt() is a no-op when the connection has already gone, so + # clear a stale flag here rather than swallow the next real close. + self._interrupted_by_sdk = False + record_environment_id(self._data_source_update_sink, action.headers) + elif isinstance(action, Event): + message_ok = False + message_handled = False + try: + message_ok = self._process_message(sink_or_store(self._data_source_update_sink, self._store), action) + message_handled = True + except json.decoder.JSONDecodeError as e: + log.info("Error while handling stream event; will restart stream: %s" % e) + self._interrupt_stream() + + if not self._handle_error(e): + break + except Exception as e: + log.info("Error while handling stream event; will restart stream: %s" % e) + self._interrupt_stream() + + if not self._handle_error(e): + break + + if message_handled: + self._retry.record_success() + + if message_ok: + self._record_stream_init(False) + self._connection_attempt_start_time = None + + if self._data_source_update_sink is not None: + self._data_source_update_sink.update_status(DataSourceState.VALID, None) + + if not self._ready.is_set(): + log.info("StreamingUpdateProcessor initialized ok.") + self._ready.set() + elif isinstance(action, Fault): + # A Fault with no error is a clean close. An interrupt the SDK + # asked for is not a failure. + if action.error is None: + if self._interrupted_by_sdk: + self._interrupted_by_sdk = False + continue + if not self._handle_error(StreamClosedError()): + break + continue + + if not self._handle_error(action.error): + break + finally: + # A raise inside the loop must not leak the connection pool. + self._sse.close() def _record_stream_init(self, failed: bool): if self._diagnostic_accumulator and self._connection_attempt_start_time: @@ -118,25 +151,34 @@ def _create_sse_client(self) -> SSEClient: url=self._uri, headers=http_factory.base_headers, pool=stream_http_factory.create_pool_manager(1, self._uri), urllib3_request_options={"timeout": stream_http_factory.timeout} ), error_strategy=ErrorStrategy.always_continue(), # we'll make error-handling decisions when we see a Fault - initial_retry_delay=self._config.initial_reconnect_delay, - retry_delay_strategy=RetryDelayStrategy.default(max_delay=MAX_RETRY_DELAY, backoff_multiplier=2, jitter_multiplier=JITTER_RATIO), - retry_delay_reset_threshold=BACKOFF_RESET_INTERVAL, + # The SSE client's retry is disabled; the SDK owns the delay. The base + # strategy must be passed: omitting it selects the library's backoff. + initial_retry_delay=0, + retry_delay_strategy=RetryDelayStrategy(), + retry_delay_reset_threshold=0, logger=log, ) def stop(self): - self.__stop_with_error_info(None) - - def __stop_with_error_info(self, error: Optional[DataSourceErrorInfo]): log.info("Stopping StreamingUpdateProcessor") self._running = False + self._stop_event.set() + + # OFF means an explicit shutdown. No stream failure produces it. It is + # reported before teardown, so a slow close cannot hold back the status + # that tells a waiter to give up. The sink drops anything after OFF. + if self._data_source_update_sink is not None: + self._data_source_update_sink.update_status(DataSourceState.OFF, None) + if self._sse: self._sse.close() - if self._data_source_update_sink is None: - return - - self._data_source_update_sink.update_status(DataSourceState.OFF, error) + def _interrupt_stream(self): + """Drops the stream connection so the next read reconnects. The SSE + client reports the close as a Fault with no error, and the flag tells + the loop that this one is ours and is already accounted for.""" + self._interrupted_by_sdk = True + self._sse.interrupt() def initialized(self): return self._running and self._ready.is_set() is True and self._store.initialized is True @@ -176,46 +218,53 @@ def _process_message(self, store, msg: Event) -> bool: # Returns true to continue, false to stop def _handle_error(self, error: Exception) -> bool: + """Records a stream failure, reports it, and waits before the retry. + + Returns True once the wait is over, or False if the processor was + stopped. No failure ever ends the stream by itself. + """ if not self._running: return False # don't retry if we've been deliberately stopped - if isinstance(error, json.decoder.JSONDecodeError): - error_info = DataSourceErrorInfo(DataSourceErrorKind.INVALID_DATA, 0, time.time(), str(error)) + self._record_stream_init(True) - log.error("Unexpected error on stream connection: %s, will retry" % error) - self._record_stream_init(True) - self._connection_attempt_start_time = None + level: Callable[..., None] - if self._data_source_update_sink is not None: - self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, error_info) + if isinstance(error, json.decoder.JSONDecodeError): + kind = FailureKind.NORMAL + error_info = DataSourceErrorInfo(DataSourceErrorKind.INVALID_DATA, 0, time.time(), str(error)) + description = "Unparseable data on stream connection: %s" % error + level = log.error elif isinstance(error, HTTPStatusError): - self._record_stream_init(True) - self._connection_attempt_start_time = None - + kind = classify_http_status(error.status) error_info = DataSourceErrorInfo(DataSourceErrorKind.ERROR_RESPONSE, error.status, time.time(), str(error)) + description = "Received %s for stream connection" % http_error_description(error.status) + level = log.error if kind is FailureKind.UNEXPECTED else log.warning + elif isinstance(error, StreamClosedError): + kind = FailureKind.NORMAL + error_info = DataSourceErrorInfo(DataSourceErrorKind.NETWORK_ERROR, 0, time.time(), str(error)) + description = "The server closed the stream connection" + level = log.warning + else: + kind = FailureKind.NORMAL + error_info = DataSourceErrorInfo(DataSourceErrorKind.UNKNOWN, 0, time.time(), str(error)) + # no stacktrace here because, for a typical connection error, it'll just be a lengthy tour of urllib3 internals + description = "Error on stream connection: %s" % error + level = log.warning - http_error_message_result = http_error_message(error.status, "stream connection") - if not is_http_error_recoverable(error.status): - log.error(http_error_message_result) - self._ready.set() # if client is initializing, make it stop waiting; has no effect if already inited - self.__stop_with_error_info(error_info) - self.stop() - return False - else: - log.warning(http_error_message_result) + self._retry.record_failure(kind) + delay = self._retry.next_delay + level("%s - will retry in %.1fs" % (description, delay)) - if self._data_source_update_sink is not None: - self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, error_info) - else: - log.warning("Unexpected error on stream connection: %s, will retry" % error) - self._record_stream_init(True) - self._connection_attempt_start_time = None + if self._data_source_update_sink is not None: + self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, error_info) - if self._data_source_update_sink is not None: - self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, DataSourceErrorInfo(DataSourceErrorKind.UNKNOWN, 0, time.time(), str(error))) - # no stacktrace here because, for a typical connection error, it'll just be a lengthy tour of urllib3 internals - self._connection_attempt_start_time = time.time() + self._sse.next_retry_delay - return True + interrupted = self._stop_event.wait(min(delay, TIMEOUT_MAX)) + + # Read after the wait, so a clock change during it cannot skew the + # stream-init latency we report. + self._connection_attempt_start_time = time.time() + return not interrupted # magic methods for "with" statement (used in testing) def __enter__(self): diff --git a/ldclient/impl/repeating_task.py b/ldclient/impl/repeating_task.py index d2e4abe1..c8c1fd62 100644 --- a/ldclient/impl/repeating_task.py +++ b/ldclient/impl/repeating_task.py @@ -1,4 +1,4 @@ -from threading import Event, Thread +from threading import TIMEOUT_MAX, Event, Thread from typing import Any, Callable from ldclient.impl.delay import DelaySource, FixedDelay @@ -66,7 +66,7 @@ def stop(self): def _run(self): if self.__initial_delay > 0: - if self.__stop.wait(self.__initial_delay): + if self.__stop.wait(min(self.__initial_delay, TIMEOUT_MAX)): return stopped = self.__stop.is_set() while not stopped: @@ -77,4 +77,4 @@ def _run(self): # The wait starts when the callback returns, so a slow callback # never shortens it. delay = self.__delays.next_delay - stopped = self.__stop.wait(delay) if delay > 0 else self.__stop.is_set() + stopped = self.__stop.wait(min(delay, TIMEOUT_MAX)) if delay > 0 else self.__stop.is_set() diff --git a/ldclient/impl/retry.py b/ldclient/impl/retry.py index a8bc0547..edfd1b85 100644 --- a/ldclient/impl/retry.py +++ b/ldclient/impl/retry.py @@ -24,13 +24,14 @@ from enum import Enum from typing import Optional, Protocol +from ldclient.config import ( + DEFAULT_INITIAL_RECONNECT_DELAY, + DEFAULT_POLL_INTERVAL +) from ldclient.impl.util import log -# The documented defaults, in seconds. Each stands in for a configured value -# that is not a positive, finite number. -DEFAULT_STREAMING_INITIAL_RECONNECT_DELAY = 1 +# The delay ceiling of the normal regime for streaming, in seconds. NORMAL_STREAMING_CEILING_DELAY = 30 -DEFAULT_POLL_INTERVAL = 30 # The delay bounds of the extended regime, in seconds. A component enters the # extended regime after an unexpected failure. @@ -53,6 +54,22 @@ _MAX_BACKOFF_EXPONENT = 30 +def _usable_delay(value: float, default: float, name: str) -> float: + """ + Returns the delay to use. A value that is not a positive, finite number of + seconds is replaced by the default. + + :param value: the configured number of seconds + :param default: the value to use when ``value`` is not usable + :param name: the option name, for the warning message + """ + + if value > 0 and math.isfinite(value): + return value + log.warning("%s must be a positive, finite number of seconds; using the default of %ss" % (name, default)) + return default + + class FailureKind(Enum): """How a failure is classified, which decides how long the next wait is.""" @@ -247,36 +264,16 @@ def _reset_if_due(self) -> None: self._max_delay = max(self._normal_ceiling_delay, self._normal_initial_delay) -def _positive_finite(value: float, default: float, name: str) -> float: - """Returns ``value`` if it is a positive, finite number of seconds, and the - default otherwise. A non-finite value would make the jitter arithmetic - produce a NaN delay, and a non-positive one would retry with no wait.""" - if value > 0 and math.isfinite(value): - return value - log.warning( - "%s must be a positive, finite number of seconds; using the default of %ss" - % (name, default) - ) - return default - - def for_streaming(initial_reconnect_delay: float) -> RetryState: """ Builds the retry state for a streaming data source. - Streaming's operating cadence is zero, so there is no delay during - healthy operation. Stream failures use either the normal or extended - initial delay to determine their backoff wait. A stream returns to - healthy operation after establishing a successful connection with no - failures during the ``STREAMING_RESET_INTERVAL``. - - ``Config`` does not check the configured delay, so the documented default - stands in for anything that is not a positive, finite number. - - The extended regime never starts below the configured delay. + Streaming's cadence is zero, so a healthy stream never waits. An invalid + delay value is replaced by the documented default; one longer than a + ceiling raises that bound rather than being cut down to it. """ - initial_reconnect_delay = _positive_finite( - initial_reconnect_delay, DEFAULT_STREAMING_INITIAL_RECONNECT_DELAY, 'initial_reconnect_delay' + initial_reconnect_delay = _usable_delay( + initial_reconnect_delay, DEFAULT_INITIAL_RECONNECT_DELAY, 'initial_reconnect_delay' ) return RetryState( normal_initial_delay=initial_reconnect_delay, @@ -292,16 +289,12 @@ def for_polling(poll_interval: float) -> RetryState: """ Builds the retry state for a polling data source. - The poll interval is polling's operating cadence, so no wait is ever - shorter than it. In the normal regime the delay bounds are the poll - interval itself, which means a normal failure simply polls again on - schedule. Polling is healthy on any successful poll, and resets after two - in a row. - - ``Config`` clamps the poll interval, but the documented default stands in - for anything that reaches here and is not a positive, finite number. + The poll interval is polling's cadence and its normal ceiling, so a normal + failure waits the interval rather than backing off past it. An invalid + interval is replaced by the documented default. No wait is ever shorter + than the interval, so the cadence wins over the extended ceiling. """ - poll_interval = _positive_finite(poll_interval, DEFAULT_POLL_INTERVAL, 'poll_interval') + poll_interval = _usable_delay(poll_interval, DEFAULT_POLL_INTERVAL, 'poll_interval') return RetryState( normal_initial_delay=poll_interval, normal_ceiling_delay=poll_interval, diff --git a/ldclient/impl/util.py b/ldclient/impl/util.py index 69e7186a..b7d1713c 100644 --- a/ldclient/impl/util.py +++ b/ldclient/impl/util.py @@ -134,6 +134,9 @@ def throw_if_unsuccessful_response(resp): def is_http_error_recoverable(status): + """ + Deprecated. Use :func:`ldclient.impl.retry.classify_http_status` instead. + """ if status >= 400 and status < 500: return status in _RETRYABLE_STATUSES # all other 4xx besides these are unrecoverable return True # all other errors are recoverable @@ -144,6 +147,10 @@ def http_error_description(status): def http_error_message(status, context, retryable_message="will retry"): + """ + Deprecated. The FDv1 data sources build their own message, so that it can + report the real retry delay. + """ return "Received %s for %s - %s" % (http_error_description(status), context, retryable_message if is_http_error_recoverable(status) else "giving up permanently") diff --git a/ldclient/interfaces.py b/ldclient/interfaces.py index 2c9c245f..de403354 100644 --- a/ldclient/interfaces.py +++ b/ldclient/interfaces.py @@ -1005,16 +1005,20 @@ class DataSourceState(Enum): In streaming mode, this means that the stream connection failed, or had to be dropped due to some other error, and will be retried after a backoff delay. In polling mode, it means that the last poll - request failed, and a new poll request will be made after the configured polling interval. + request failed, and a new poll request will be made after the polling interval, or after a longer + delay if the error is one that needs to be fixed. """ OFF = 'off' """ - Indicates that the data source has been permanently shut down. + Indicates that the data source is permanently shut down. - This could be because it encountered an unrecoverable error (for instance, the LaunchDarkly service - rejected the SDK key; an invalid SDK key will never become valid), or because the SDK client was - explicitly shut down. + This could be because the SDK client was explicitly shut down, because its configuration could not + be parsed, or because the data source encountered a condition it will not retry. + + No further state or data follows. A request or connection that was still in flight when the data + source stopped is not reported, so this state is final for the lifetime of the data source. It is + reported when the shutdown begins rather than when the last connection closes. """ diff --git a/ldclient/testing/impl/datasource/test_async_polling.py b/ldclient/testing/impl/datasource/test_async_polling.py index 919bf944..8d497ae0 100644 --- a/ldclient/testing/impl/datasource/test_async_polling.py +++ b/ldclient/testing/impl/datasource/test_async_polling.py @@ -3,9 +3,14 @@ """ import asyncio +import logging +import ssl +import time from unittest.mock import AsyncMock, MagicMock, patch +import aiohttp import pytest +from aiohttp.client_reqrep import ConnectionKey from ldclient.config import Config from ldclient.impl.aio.transport_types import TransportResponse @@ -13,6 +18,12 @@ AsyncFeatureRequesterImpl ) from ldclient.impl.datasource.async_polling import AsyncPollingUpdateProcessor +from ldclient.impl.retry import ( + POLLING_RESET_SUCCESSES, + AfterConsecutiveSuccesses, + RetryState, + for_polling +) from ldclient.impl.util import UnsuccessfulResponseException from ldclient.interfaces import ( AsyncDataSourceUpdateSink, @@ -20,6 +31,7 @@ DataSourceState ) from ldclient.testing.mock_async_components import MockAsyncFeatureStore +from ldclient.testing.test_util import no_retry_jitter from ldclient.versioned_data_kind import FEATURES, SEGMENTS # Sample data returned by a successful poll @@ -33,7 +45,37 @@ def make_config(**kwargs): return Config('SDK_KEY', **kwargs) -def make_processor(config=None, store=None, ready=None, requester=None): +# aiohttp's connection errors read the connection key when they are turned +# into a string, which the data source does, so a real one is needed here. +_CONNECTION_KEY = ConnectionKey( + host='app.launchdarkly.com', + port=443, + is_ssl=True, + ssl=True, + proxy=None, + proxy_auth=None, + proxy_headers_hash=None, + server_hostname=None, +) + + +ONE_HOUR = 60 * 60 + + +def fast_retry_state(delay=0.001): + """A retry state whose every delay is ``delay``: small enough to skip the + real extended-regime wait, or large enough to prove a stop interrupts one.""" + return RetryState( + normal_initial_delay=delay, + normal_ceiling_delay=delay, + extended_initial_delay=delay, + extended_ceiling_delay=delay, + reset_policy=AfterConsecutiveSuccesses(POLLING_RESET_SUCCESSES), + operating_cadence=delay, + ) + + +def make_processor(config=None, store=None, ready=None, requester=None, retry_state=None): if config is None: config = make_config() if store is None: @@ -48,6 +90,7 @@ def make_processor(config=None, store=None, ready=None, requester=None): requester=requester, store=store, ready=ready, + retry_state=retry_state, ) @@ -174,31 +217,69 @@ async def test_successful_poll_initializes_store_and_sets_ready(self, mock_inter @pytest.mark.asyncio @patch('ldclient.config.Config.poll_interval', new_callable=MagicMock) - async def test_unrecoverable_http_error_stops_polling_and_sets_ready(self, mock_interval): + async def test_unexpected_http_error_keeps_polling_and_leaves_ready_unset(self, mock_interval): mock_interval.__get__ = MagicMock(return_value=0) store = MockAsyncFeatureStore() ready = asyncio.Event() config = make_config() - processor = make_processor(config=config, store=store, ready=ready) + processor = make_processor(config=config, store=store, ready=ready, retry_state=fast_retry_state()) mock_requester = AsyncMock(side_effect=UnsuccessfulResponseException(401)) processor._requester.get_all_data = mock_requester processor.start() - await asyncio.wait_for(ready.wait(), timeout=2.0) + await asyncio.sleep(0.1) - assert ready.is_set() + # A rejected SDK key must not falsely unblock initialization, and it + # must not stop the poller. + assert not ready.is_set() assert not processor.initialized() + assert mock_requester.call_count >= 2 - # The polling task must have stopped itself: no further polls occur. - await asyncio.sleep(0.05) - snapshot = mock_requester.call_count + await processor.stop() + + @pytest.mark.asyncio + @patch('ldclient.config.Config.poll_interval', new_callable=MagicMock) + async def test_unexpected_http_error_moves_to_the_extended_regime(self, mock_interval): + mock_interval.__get__ = MagicMock(return_value=0) + + retry = fast_retry_state() + processor = make_processor(retry_state=retry) + processor._requester.get_all_data = AsyncMock(side_effect=UnsuccessfulResponseException(401)) + + processor.start() await asyncio.sleep(0.05) - assert mock_requester.call_count == snapshot + + assert retry._extended await processor.stop() + @pytest.mark.asyncio + async def test_the_first_success_after_an_outage_polls_at_the_cadence(self): + # A backoff wait applies to a retry, not to every operation. The + # retry state carries the wait, so this reads it there rather than + # measuring elapsed time. + store = MockAsyncFeatureStore() + ready = asyncio.Event() + config = make_config() + with no_retry_jitter(): + retry = for_polling(30) + processor = make_processor(config=config, store=store, ready=ready, retry_state=retry) + + processor._requester.get_all_data = AsyncMock(side_effect=UnsuccessfulResponseException(401)) + await processor._fetch_and_store() + assert retry.next_delay == 5 * 60 + + processor._requester.get_all_data = AsyncMock(return_value=SAMPLE_DATA) + await processor._fetch_and_store() + assert retry.next_delay == 30 + assert retry._extended, "one success restores the cadence but does not reset" + + await processor._fetch_and_store() + assert retry.next_delay == 30 + assert not retry._extended, "two successes in a row reset the state" + @pytest.mark.asyncio @patch('ldclient.config.Config.poll_interval', new_callable=MagicMock) async def test_recoverable_http_error_continues_polling(self, mock_interval): @@ -207,7 +288,7 @@ async def test_recoverable_http_error_continues_polling(self, mock_interval): store = MockAsyncFeatureStore() ready = asyncio.Event() config = make_config() - processor = make_processor(config=config, store=store, ready=ready) + processor = make_processor(config=config, store=store, ready=ready, retry_state=fast_retry_state()) call_count = 0 @@ -237,7 +318,7 @@ async def test_general_exception_does_not_stop_polling(self, mock_interval): store = MockAsyncFeatureStore() ready = asyncio.Event() config = make_config() - processor = make_processor(config=config, store=store, ready=ready) + processor = make_processor(config=config, store=store, ready=ready, retry_state=fast_retry_state()) call_count = 0 @@ -291,6 +372,69 @@ async def get_all_data(): await processor.stop() + @pytest.mark.parametrize( + "error", + [ + aiohttp.ClientConnectorCertificateError( + _CONNECTION_KEY, ssl.SSLCertVerificationError("self-signed certificate") + ), + aiohttp.ClientConnectorSSLError(_CONNECTION_KEY, OSError("handshake failed")), + ssl.SSLEOFError("EOF occurred in violation of protocol"), + ConnectionResetError(104, "reset by peer"), + ], + ids=["aiohttp-certificate", "aiohttp-tls", "peer-close-handshake", "reset"], + ) + @pytest.mark.asyncio + async def test_transport_failures_poll_again_at_the_cadence(self, error): + """No transport failure reaches the extended regime, an aiohttp + certificate failure included. Only an HTTP status can do that.""" + retry = for_polling(30) + processor = make_processor(retry_state=retry) + processor._requester.get_all_data = AsyncMock(side_effect=error) + + await processor._fetch_and_store() + assert retry.next_delay == 30 + assert not retry._extended + + @pytest.mark.asyncio + async def test_the_log_reports_the_growing_retry_delay(self, caplog): + """The message has to carry the real delay, so someone reading logs can + see the backoff working.""" + caplog.set_level(logging.WARNING) + + with no_retry_jitter(): + retry = for_polling(30) + processor = make_processor(retry_state=retry) + processor._requester.get_all_data = AsyncMock(side_effect=UnsuccessfulResponseException(401)) + + await processor._fetch_and_store() + await processor._fetch_and_store() + + messages = [r.getMessage() for r in caplog.records] + assert messages == [ + "Received HTTP error 401 (invalid SDK key) for polling request - will retry in 300.0s", + "Received HTTP error 401 (invalid SDK key) for polling request - will retry in 600.0s", + ] + # An error a person has to fix is logged at error level, every time. + assert [r.levelno for r in caplog.records] == [logging.ERROR, logging.ERROR] + + @pytest.mark.asyncio + async def test_a_transport_error_reports_a_delay_and_keeps_its_stacktrace(self, caplog): + caplog.set_level(logging.WARNING) + + with no_retry_jitter(): + retry = for_polling(30) + processor = make_processor(retry_state=retry) + processor._requester.get_all_data = AsyncMock(side_effect=ConnectionResetError(104, "reset by peer")) + + await processor._fetch_and_store() + + record = caplog.records[0] + assert record.getMessage() == "Error encountered when updating flags: [Errno 104] reset by peer - will retry in 30.0s" + # The handler has exited by the time this is logged, so the exception + # has to be carried explicitly for the traceback to survive. + assert record.exc_info is not None + @pytest.mark.asyncio async def test_stop_closes_requester(self): processor = make_processor() @@ -329,6 +473,30 @@ async def close(): assert order == ['poll_done', 'transport_closed'] + @pytest.mark.asyncio + async def test_an_extended_regime_wait_is_cut_short_by_stop(self): + """Shutdown must not sit through an hour-long backoff. The in-flight-poll + case is test_stop_cancels_polling_task_cleanly; this one stops while the + task is waiting between polls.""" + retry = fast_retry_state(ONE_HOUR) + processor = make_processor(retry_state=retry) + processor._requester.get_all_data = AsyncMock( + side_effect=UnsuccessfulResponseException(401) + ) + + processor.start() + # Confirm the wait under test really is long before measuring the stop. + deadline = time.time() + 2 + while retry.next_delay <= 60 and time.time() < deadline: + await asyncio.sleep(0.01) + assert retry.next_delay > 60, "the wait under test should be minutes long" + + started = time.time() + await processor.stop() + elapsed = time.time() - started + + assert elapsed < 2, "stop() took %.2fs" % elapsed + @pytest.mark.asyncio async def test_stop_cancels_polling_task_cleanly(self): store = MockAsyncFeatureStore() @@ -357,7 +525,7 @@ async def slow_poll(): @pytest.mark.asyncio @patch('ldclient.config.Config.poll_interval', new_callable=MagicMock) - async def test_unrecoverable_error_updates_sink_to_off(self, mock_interval): + async def test_unexpected_error_updates_sink_to_interrupted_never_off(self, mock_interval): mock_interval.__get__ = MagicMock(return_value=0) store = MockAsyncFeatureStore() @@ -367,7 +535,7 @@ async def test_unrecoverable_error_updates_sink_to_off(self, mock_interval): sink = MagicMock(spec=AsyncDataSourceUpdateSink) config._data_source_update_sink = sink - processor = make_processor(config=config, store=store, ready=ready) + processor = make_processor(config=config, store=store, ready=ready, retry_state=fast_retry_state()) processor._data_source_update_sink = sink processor._requester.get_all_data = AsyncMock( @@ -375,15 +543,89 @@ async def test_unrecoverable_error_updates_sink_to_off(self, mock_interval): ) processor.start() - await asyncio.wait_for(ready.wait(), timeout=2.0) + await asyncio.sleep(0.05) - # Verify the sink was told to go OFF - calls = [call for call in sink.update_status.call_args_list if call.args[0] == DataSourceState.OFF] - assert len(calls) >= 1 - error_info = calls[0].args[1] + interrupted = [c for c in sink.update_status.call_args_list if c.args[0] == DataSourceState.INTERRUPTED] + assert len(interrupted) >= 1 + error_info = interrupted[0].args[1] assert error_info.kind == DataSourceErrorKind.ERROR_RESPONSE assert error_info.status_code == 403 + assert not any(c.args[0] == DataSourceState.OFF for c in sink.update_status.call_args_list) + + await processor.stop() + + @pytest.mark.asyncio + @patch('ldclient.config.Config.poll_interval', new_callable=MagicMock) + async def test_stop_updates_sink_to_off(self, mock_interval): + mock_interval.__get__ = MagicMock(return_value=0) + + config = make_config() + sink = MagicMock(spec=AsyncDataSourceUpdateSink) + config._data_source_update_sink = sink + + processor = make_processor(config=config) + processor._data_source_update_sink = sink + processor._requester.get_all_data = AsyncMock(return_value=SAMPLE_DATA) + + processor.start() + await processor.stop() + + assert any(c.args[0] == DataSourceState.OFF for c in sink.update_status.call_args_list) + + @pytest.mark.asyncio + async def test_a_poll_finishing_after_stop_reports_nothing(self): + """The poll still in flight when stop() ran must not report after OFF.""" + from ldclient.impl.datasource.async_status import ( + AsyncDataSourceUpdateSinkImpl + ) + from ldclient.impl.listeners import Listeners + + store = MockAsyncFeatureStore() + observed = [] + listeners = Listeners() + listeners.add(lambda status: observed.append(status.state)) + + config = make_config() + config._data_source_update_sink = AsyncDataSourceUpdateSinkImpl(store, listeners, Listeners()) + + processor = make_processor(config=config, store=store) + processor._requester.get_all_data = AsyncMock(return_value=SAMPLE_DATA) + + await processor.stop() + await processor._fetch_and_store() + + assert observed == [DataSourceState.OFF] + + @pytest.mark.asyncio + @patch('ldclient.config.Config.poll_interval', new_callable=MagicMock) + async def test_valid_status_is_reported_before_ready_is_set(self, mock_interval): + # Mirrors go-server-sdk#442: a caller that wakes on readiness must not + # still be able to read INITIALIZING. + mock_interval.__get__ = MagicMock(return_value=0) + + from ldclient.impl.datasource.async_status import ( + AsyncDataSourceUpdateSinkImpl + ) + from ldclient.impl.listeners import Listeners + + store = MockAsyncFeatureStore() + ready = asyncio.Event() + observed = [] + listeners = Listeners() + listeners.add(lambda status: observed.append((status.state, ready.is_set()))) + + config = make_config() + config._data_source_update_sink = AsyncDataSourceUpdateSinkImpl(store, listeners, Listeners()) + + processor = make_processor(config=config, store=store, ready=ready) + processor._requester.get_all_data = AsyncMock(return_value=SAMPLE_DATA) + + processor.start() + await asyncio.wait_for(ready.wait(), timeout=2.0) + + assert observed[0] == (DataSourceState.VALID, False) + await processor.stop() @pytest.mark.asyncio @@ -449,15 +691,14 @@ async def test_stop_closes_transport_when_cancelled_mid_wait(self): requester.close = AsyncMock() processor = make_processor(requester=requester) - # Replace the repeating task so wait_stopped() hangs until we cancel stop(). + # Replace the task's wait so it hangs until we cancel stop(). waiting = asyncio.Event() async def hang(): waiting.set() await asyncio.Event().wait() - processor._task = MagicMock() - processor._task.wait_stopped = hang + processor._task.wait_stopped = hang # type: ignore[method-assign] stop_task = asyncio.create_task(processor.stop()) await asyncio.wait_for(waiting.wait(), timeout=2.0) diff --git a/ldclient/testing/impl/datasource/test_async_status.py b/ldclient/testing/impl/datasource/test_async_status.py index 6790a922..25b8e635 100644 --- a/ldclient/testing/impl/datasource/test_async_status.py +++ b/ldclient/testing/impl/datasource/test_async_status.py @@ -333,3 +333,35 @@ async def test_upsert_records_store_error_on_failure(): assert len(status_capture.statuses) == 1 assert status_capture.statuses[0].error.kind == DataSourceErrorKind.STORE_ERROR + + +@pytest.mark.asyncio +async def test_update_status_off_is_terminal(): + sink, status_listeners, _ = make_sink() + status_capture = StatusCapture() + status_listeners.add(status_capture) + + sink.update_status(DataSourceState.VALID, None) + sink.update_status(DataSourceState.OFF, None) + + # A poll or stream connection still in flight when the data source stopped. + sink.update_status(DataSourceState.VALID, None) + sink.update_status(DataSourceState.INTERRUPTED, DataSourceErrorInfo(DataSourceErrorKind.NETWORK_ERROR, 0, 1000, 'late')) + + assert sink.status.state == DataSourceState.OFF + assert sink.status.error is None + assert [status.state for status in status_capture.statuses] == [DataSourceState.VALID, DataSourceState.OFF] + + +@pytest.mark.asyncio +async def test_store_error_after_off_reports_nothing(): + sink, status_listeners, _ = make_sink(_FailingStore()) + status_capture = StatusCapture() + status_listeners.add(status_capture) + sink.update_status(DataSourceState.OFF, None) + + with pytest.raises(RuntimeError): + await sink.upsert(FEATURES, make_flag('flag-a').to_json_dict()) + + assert sink.status.state == DataSourceState.OFF + assert [status.state for status in status_capture.statuses] == [DataSourceState.OFF] diff --git a/ldclient/testing/impl/datasource/test_async_streaming.py b/ldclient/testing/impl/datasource/test_async_streaming.py index ffd5a9ea..0c8865c3 100644 --- a/ldclient/testing/impl/datasource/test_async_streaming.py +++ b/ldclient/testing/impl/datasource/test_async_streaming.py @@ -7,9 +7,14 @@ import asyncio import json +import logging +import ssl +import time from unittest import mock +import aiohttp import pytest +from aiohttp.client_reqrep import ConnectionKey from ldclient.config import Config from ldclient.impl.datasource import async_streaming @@ -17,9 +22,23 @@ AsyncStreamingUpdateProcessor ) from ldclient.impl.model import ModelEntity +from ldclient.impl.retry import ( + EXTENDED_CEILING_DELAY, + EXTENDED_INITIAL_DELAY, + NORMAL_STREAMING_CEILING_DELAY, + STREAMING_RESET_INTERVAL, + AfterHealthyFor, + RetryState, + for_streaming +) from ldclient.interfaces import DataSourceErrorKind, DataSourceState from ldclient.testing.builders import FlagBuilder, SegmentBuilder from ldclient.testing.mock_async_components import MockAsyncFeatureStore +from ldclient.testing.test_util import ( + no_retry_jitter, + record_healthy_windows, + ticking_clock +) from ldclient.versioned_data_kind import FEATURES, SEGMENTS @@ -75,6 +94,60 @@ async def _actions_generator(actions: list): await asyncio.Event().wait() +def _retry_state_with(policy: AfterHealthyFor) -> RetryState: + """A retry state with tiny delays and a caller-supplied reset policy, so a + test can watch the window.""" + return RetryState( + normal_initial_delay=0.001, + normal_ceiling_delay=0.001, + extended_initial_delay=0.001, + extended_ceiling_delay=0.001, + reset_policy=policy, + ) + + +ONE_HOUR = 60 * 60 + + +def _fast_retry_state(delay: float = 0.001) -> RetryState: + """A retry state whose every delay is ``delay``: small enough to skip the + real extended-regime wait, or large enough to prove a stop interrupts one.""" + return RetryState( + normal_initial_delay=delay, + normal_ceiling_delay=delay, + extended_initial_delay=delay, + extended_ceiling_delay=delay, + reset_policy=AfterHealthyFor(STREAMING_RESET_INTERVAL), + ) + + +# aiohttp's connection errors read the connection key when they are turned +# into a string, which the data source does, so a real one is needed here. +_CONNECTION_KEY = ConnectionKey( + host='stream.launchdarkly.com', + port=443, + is_ssl=True, + ssl=True, + proxy=None, + proxy_auth=None, + proxy_headers_hash=None, + server_hostname=None, +) + + +def _zero_delay_retry_state() -> RetryState: + """A retry state whose normal regime waits no time at all, so a test can + drive ``_handle_error`` without a real sleep. The extended bounds stay + real, so a misclassification still shows up in ``max_delay``.""" + return RetryState( + normal_initial_delay=0, + normal_ceiling_delay=NORMAL_STREAMING_CEILING_DELAY, + extended_initial_delay=EXTENDED_INITIAL_DELAY, + extended_ceiling_delay=EXTENDED_CEILING_DELAY, + reset_policy=AfterHealthyFor(STREAMING_RESET_INTERVAL), + ) + + class _MockSSE: """Stand-in for AsyncSSEClient exposing the surface the processor uses.""" @@ -101,31 +174,33 @@ class _MockSSEFactory: def __init__(self, actions: list): self._actions = actions self.created: list = [] + self.sdk_managed_retry: list = [] - def create(self, url: str, initial_retry_delay: float) -> _MockSSE: + def create(self, url: str, initial_retry_delay: float, sdk_managed_retry: bool = False) -> _MockSSE: sse = _MockSSE(self._actions) self.created.append(sse) + self.sdk_managed_retry.append(sdk_managed_retry) return sse -def _make_processor(actions, config=None, store=None, ready_event=None, diag=None): +def _make_processor(actions, config=None, store=None, ready_event=None, diag=None, retry_state=None): config = config or _make_config() store = store or MockAsyncFeatureStore() ready_event = ready_event or asyncio.Event() factory = _MockSSEFactory(actions) - proc = AsyncStreamingUpdateProcessor(config, store, ready_event, diag, factory) + proc = AsyncStreamingUpdateProcessor(config, store, ready_event, diag, factory, retry_state=retry_state) return proc, store, ready_event, factory async def _run_with_actions(actions: list, config=None, store=None, ready_event=None, - diag=None, extra_ready_timeout=3.0): + diag=None, extra_ready_timeout=3.0, retry_state=None): """Run the processor against a fake SSE action sequence. Starts the processor and waits for the ready event (up to *extra_ready_timeout* seconds), then returns ``(processor, store, ready_event, factory)``. """ - proc, store, ready, factory = _make_processor(actions, config, store, ready_event, diag) + proc, store, ready, factory = _make_processor(actions, config, store, ready_event, diag, retry_state) proc.start() try: await asyncio.wait_for(ready.wait(), timeout=extra_ready_timeout) @@ -223,27 +298,135 @@ async def test_fault_with_error_does_not_set_ready_by_itself(): @pytest.mark.asyncio -async def test_fault_none_error_is_ignored(): - """A Fault with error=None (clean close) should not update status or stop the processor.""" +async def test_server_close_backs_off_and_does_not_stop_the_processor(): + """A Fault with error=None is the server closing a connection it normally + leaves open. The SDK backs off rather than reconnecting at once, but the + processor keeps running.""" flag = FlagBuilder('f1').version(1).build() put_data = _make_put_data(flags={'f1': _item_dict(flag)}) actions = [ _start(), _event('put', put_data), - _fault(error=None), # clean close — should be ignored + _fault(error=None), # clean close by the server ] - proc, store, ready, _ = await _run_with_actions(actions) + retry = _fast_retry_state() + proc, store, ready, factory = _make_processor(actions, retry_state=retry) + proc.start() + await asyncio.wait_for(ready.wait(), timeout=3.0) + await _wait_until(lambda: retry._attempts >= 1) - assert ready.is_set() assert store.initialized + assert not factory.created[0].closed + assert not retry._extended + + await proc.stop() + + +@pytest.mark.asyncio +async def test_server_close_reports_a_network_error(): + from ldclient.impl.datasource.async_status import ( + AsyncDataSourceUpdateSinkImpl + ) + from ldclient.impl.listeners import Listeners + + store = MockAsyncFeatureStore() + statuses = [] + listeners = Listeners() + listeners.add(lambda s: statuses.append(s)) + + config = _make_config() + config._data_source_update_sink = AsyncDataSourceUpdateSinkImpl(store, listeners, Listeners()) + + flag = FlagBuilder('f1').version(1).build() + put_data = _make_put_data(flags={'f1': _item_dict(flag)}) + actions = [_start(), _event('put', put_data), _fault(error=None)] + + proc, store, ready, _ = _make_processor(actions, config=config, store=store, retry_state=_fast_retry_state()) + proc.start() + await _wait_until(lambda: any(s.state == DataSourceState.INTERRUPTED for s in statuses)) + + interrupted = [s for s in statuses if s.state == DataSourceState.INTERRUPTED] + assert interrupted[0].error is not None + assert interrupted[0].error.kind == DataSourceErrorKind.NETWORK_ERROR + + await proc.stop() + + +@pytest.mark.asyncio +async def test_repeated_server_closes_stay_on_the_normal_curve(): + """A load balancer draining during a rolling deploy closes streams + cleanly, over and over. That must never reach the extended regime.""" + flag = FlagBuilder('f1').version(1).build() + put_data = _make_put_data(flags={'f1': _item_dict(flag)}) + actions = [] + for _ in range(10): + actions += [_start(), _event('put', put_data), _fault(error=None)] + + retry = _fast_retry_state() + proc, store, ready, _ = _make_processor(actions, retry_state=retry) + proc.start() + await _wait_until(lambda: retry._attempts >= 10, timeout=5.0) + + assert not retry._extended + assert retry._max_delay == _fast_retry_state()._max_delay + + await proc.stop() + + +@pytest.mark.asyncio +async def test_our_own_interrupt_is_not_counted_as_a_server_close(): + """Bad JSON makes the SDK drop the connection itself. The SSE client then + reports that close as a Fault with no error, and counting it would record + the same failure twice and wait twice.""" + flag = FlagBuilder('f1').version(1).build() + put_data = _make_put_data(flags={'f1': _item_dict(flag)}) + actions = [ + _start(), + _event('put', put_data), + _event('patch', 'not valid json'), + _fault(error=None), # the close our own interrupt caused + ] + + retry = _fast_retry_state() + proc, store, ready, _ = _make_processor(actions, retry_state=retry) + proc.start() + await _wait_until(lambda: retry._attempts >= 1) + await asyncio.sleep(0.1) + + assert retry._attempts == 1 + + await proc.stop() + + +@pytest.mark.asyncio +async def test_a_leaked_interrupt_flag_does_not_swallow_a_server_close(): + """interrupt() is a no-op when the connection has already gone, so no + Fault arrives to clear the flag. A new connection must clear it, or the + next genuine close is recorded as ours and the backoff is skipped.""" + put_data = _make_put_data() + actions = [ + _start(), + _event('put', put_data), + _fault(error=None), # a close the SDK did not ask for + ] + + retry = _fast_retry_state() + proc, _, _, _ = _make_processor(actions, retry_state=retry) + proc._interrupted_by_sdk = True + proc.start() + await _wait_until(lambda: retry._attempts >= 1) + await asyncio.sleep(0.1) + + assert retry._attempts == 1 await proc.stop() @pytest.mark.asyncio -async def test_unrecoverable_http_error_stops_processor(): - """An unrecoverable HTTP status closes the stream and reports OFF with error info.""" +async def test_unexpected_http_error_keeps_the_processor_running(): + """A rejected SDK key is retried like any other failure. The state never + goes OFF, and initialization is not falsely unblocked.""" from ld_eventsource.errors import HTTPStatusError from ldclient.impl.datasource.async_status import ( @@ -261,15 +444,17 @@ async def test_unrecoverable_http_error_stops_processor(): actions = [_start(), _fault(error=HTTPStatusError(401))] - proc, store, ready, factory = await _run_with_actions(actions, config=config, store=store) + proc, store, ready, factory = await _run_with_actions( + actions, config=config, store=store, extra_ready_timeout=0.2, + retry_state=_fast_retry_state(), + ) - # The unrecoverable error unblocks initialization without initializing the store. - assert ready.is_set() + assert not ready.is_set() assert not proc.initialized() - assert factory.created[0].closed + assert not factory.created[0].closed + assert all(s.state != DataSourceState.OFF for s in statuses) assert any( - s.state == DataSourceState.OFF - and s.error is not None + s.error is not None and s.error.kind == DataSourceErrorKind.ERROR_RESPONSE and s.error.status_code == 401 for s in statuses @@ -278,6 +463,197 @@ async def test_unrecoverable_http_error_stops_processor(): await proc.stop() +@pytest.mark.asyncio +async def test_unexpected_http_error_moves_to_the_extended_regime(): + from ld_eventsource.errors import HTTPStatusError + + actions = [_start(), _fault(error=HTTPStatusError(401))] + retry = _fast_retry_state() + proc, store, ready, _ = _make_processor(actions, retry_state=retry) + proc.start() + await _wait_until(lambda: retry._extended) + + await proc.stop() + + +@pytest.mark.asyncio +async def test_normal_http_error_stays_in_the_normal_regime(): + from ld_eventsource.errors import HTTPStatusError + + actions = [_start(), _fault(error=HTTPStatusError(503))] + retry = _fast_retry_state() + proc, store, ready, _ = _make_processor(actions, retry_state=retry) + proc.start() + await _wait_until(lambda: retry._attempts >= 1) + + assert not retry._extended + + await proc.stop() + + +@pytest.mark.parametrize( + "error", + [ + aiohttp.ClientConnectorCertificateError( + _CONNECTION_KEY, ssl.SSLCertVerificationError("self-signed certificate") + ), + aiohttp.ClientConnectorSSLError(_CONNECTION_KEY, OSError("handshake failed")), + ssl.SSLEOFError("EOF occurred in violation of protocol"), + ConnectionResetError(104, "reset by peer"), + ], + ids=["aiohttp-certificate", "aiohttp-tls", "peer-close-handshake", "reset"], +) +@pytest.mark.asyncio +async def test_transport_failures_stay_in_the_normal_regime(error): + """No transport failure reaches the extended regime, an aiohttp + certificate failure included. Only an HTTP status can do that.""" + retry = _zero_delay_retry_state() + proc, store, ready, _ = _make_processor([], retry_state=retry) + proc._running = True + + # A misclassification would wait five minutes here, so bound the wait + # rather than let the test hang. + assert await asyncio.wait_for(proc._handle_error(error), timeout=2.0) + + assert not retry._extended + assert retry._max_delay == NORMAL_STREAMING_CEILING_DELAY + + +class _NoSleep: + """Stands in for the ``asyncio`` module inside async_streaming, so the wait + in _handle_error returns at once. ``sleep`` is all that module uses.""" + + def __init__(self): + self.slept: list = [] + + async def sleep(self, seconds): + self.slept.append(seconds) + + +@pytest.mark.asyncio +async def test_the_log_reports_the_growing_retry_delay(caplog): + """The message has to carry the real delay, so someone reading logs can see + the backoff working. The vaguer wording it replaced could not show this.""" + from ld_eventsource.errors import HTTPStatusError + + caplog.set_level(logging.WARNING) + no_sleep = _NoSleep() + + with no_retry_jitter(), mock.patch.object(async_streaming, 'asyncio', no_sleep): + retry = for_streaming(1) + proc, store, ready, _ = _make_processor([], retry_state=retry) + proc._running = True + + await proc._handle_error(HTTPStatusError(401)) + await proc._handle_error(HTTPStatusError(401)) + + messages = [r.getMessage() for r in caplog.records] + assert messages == [ + "Received HTTP error 401 (invalid SDK key) for stream connection - will retry in 300.0s", + "Received HTTP error 401 (invalid SDK key) for stream connection - will retry in 600.0s", + ] + # An error a person has to fix is logged at error level, every time. + assert [r.levelno for r in caplog.records] == [logging.ERROR, logging.ERROR] + # The reported delay is the one actually waited. + assert no_sleep.slept == [5 * 60, 10 * 60] + + +@pytest.mark.asyncio +async def test_the_processor_asks_the_factory_to_leave_the_delay_to_the_sdk(): + proc, store, ready, factory = _make_processor([]) + proc.start() + await _wait_until(lambda: len(factory.created) > 0) + + assert factory.sdk_managed_retry == [True] + + await proc.stop() + + +@pytest.mark.asyncio +async def test_several_messages_on_one_stream_do_not_extend_the_reset_window(): + """The window starts at the first message and stays there, however many + more arrive on the same stream.""" + flag = FlagBuilder('f1').version(1).build() + put_data = _make_put_data(flags={'f1': _item_dict(flag)}) + patch_data = _make_patch_data(FEATURES, _item_dict(FlagBuilder('f1').version(2).build())) + actions = [_start(), _event('put', put_data), _event('patch', patch_data)] + + policy = AfterHealthyFor(STREAMING_RESET_INTERVAL) + retry = _retry_state_with(policy) + # The clock moves on every read, so a window that had been restarted reads + # back as a different time. + windows = record_healthy_windows(policy) + + with ticking_clock(): + proc, store, ready, _ = _make_processor(actions, retry_state=retry) + proc.start() + await _wait_until(lambda: len(windows) >= 2) + await proc.stop() + + assert len(set(windows)) == 1, "the window moved between messages" + + +@pytest.mark.asyncio +async def test_a_fresh_stream_starts_a_new_reset_window(): + """A stream teardown clears the window through record_failure, so the next + stream measures its own stretch rather than inheriting the old one.""" + from ld_eventsource.errors import HTTPStatusError + + flag = FlagBuilder('f1').version(1).build() + put_data = _make_put_data(flags={'f1': _item_dict(flag)}) + actions = [ + _start(), + _event('put', put_data), + _fault(error=HTTPStatusError(503)), + _start(), + _event('put', put_data), + ] + + policy = AfterHealthyFor(STREAMING_RESET_INTERVAL) + retry = _retry_state_with(policy) + windows = record_healthy_windows(policy) + + with ticking_clock(): + proc, store, ready, _ = _make_processor(actions, retry_state=retry) + proc.start() + # One put per stream, so two signals in all. + await _wait_until(lambda: len(windows) >= 2) + await proc.stop() + + assert len(set(windows)) == 2, "the second stream reused the first window" + + +@pytest.mark.asyncio +async def test_an_extended_regime_wait_is_cut_short_by_stop(): + """Shutdown must not sit through an hour-long backoff. The healthy-stop case + is test_stop_closes_sse_and_finishes_task; this one stops mid-wait.""" + from ld_eventsource.errors import HTTPStatusError + + retry = _fast_retry_state(ONE_HOUR) + proc, store, ready, _ = _make_processor( + [_start(), _fault(error=HTTPStatusError(401))], retry_state=retry + ) + proc.start() + # Confirm the wait under test really is long before measuring the stop. + await _wait_until(lambda: retry.next_delay > 60) + + started = time.time() + await proc.stop() + elapsed = time.time() - started + + assert elapsed < 2, "stop() took %.2fs" % elapsed + leaked = [t for t in proc._runner._tasks if not t.done()] + assert leaked == [], "stop() returned with the task still running: %r" % leaked + + +@pytest.mark.asyncio +async def test_stop_before_start_and_stop_twice_are_safe(): + proc, store, ready, _ = _make_processor([]) + + await proc.stop() + await proc.stop() + + @pytest.mark.asyncio async def test_stop_closes_sse_and_finishes_task(): flag = FlagBuilder('f1').version(1).build() @@ -295,13 +671,16 @@ async def test_stop_closes_sse_and_finishes_task(): @pytest.mark.asyncio -async def test_second_start_raises(): +async def test_second_start_is_a_no_op(): + """AsyncLDClient.start() is documented as an idempotent no-op, so nothing + underneath it may raise on a repeat call.""" actions = [_start()] - proc, store, ready, _ = _make_processor(actions) + proc, store, ready, factory = _make_processor(actions) proc.start() try: - with pytest.raises(RuntimeError): - proc.start() + proc.start() + await _wait_until(lambda: len(factory.created) > 0) + assert len(factory.created) == 1 finally: await proc.stop() @@ -494,3 +873,38 @@ async def test_diagnostics_recorded_on_successful_init(): assert recorded[0]['failed'] is False await proc.stop() + + +@pytest.mark.asyncio +async def test_off_is_reported_before_teardown(): + """A slow close must not hold back the status that tells a waiter to give + up, so OFF goes out before the connection and session are torn down.""" + order = [] + + class _OrderingSink: + async def init(self, all_data): + pass + + def update_status(self, new_state, new_error): + order.append(new_state) + + config = _make_config() + config._data_source_update_sink = _OrderingSink() + + flag = FlagBuilder('f1').version(1).build() + put_data = _make_put_data(flags={'f1': _item_dict(flag)}) + proc, _, _, factory = await _run_with_actions([_start(), _event('put', put_data)], config=config) + + sse = factory.created[0] + real_close = sse.close + + async def close(): + order.append('closed') + await real_close() + + sse.close = close + + await proc.stop() + + assert DataSourceState.OFF in order + assert order.index(DataSourceState.OFF) < order.index('closed') diff --git a/ldclient/testing/impl/datasource/test_polling_processor.py b/ldclient/testing/impl/datasource/test_polling_processor.py index 06e92d89..2ae603ff 100644 --- a/ldclient/testing/impl/datasource/test_polling_processor.py +++ b/ldclient/testing/impl/datasource/test_polling_processor.py @@ -1,13 +1,22 @@ +import logging +import ssl import threading import time import mock +import pytest from ldclient.config import Config from ldclient.feature_store import InMemoryFeatureStore from ldclient.impl.datasource.polling import PollingUpdateProcessor from ldclient.impl.datasource.status import DataSourceUpdateSinkImpl from ldclient.impl.listeners import Listeners +from ldclient.impl.retry import ( + POLLING_RESET_SUCCESSES, + AfterConsecutiveSuccesses, + RetryState, + for_polling +) from ldclient.impl.util import UnsuccessfulResponseException from ldclient.interfaces import ( DataSourceErrorKind, @@ -16,7 +25,8 @@ ) from ldclient.testing.builders import * from ldclient.testing.stub_util import MockFeatureRequester, MockResponse -from ldclient.testing.test_util import SpyListener +from ldclient.testing.sync_util import wait_until +from ldclient.testing.test_util import SpyListener, no_retry_jitter from ldclient.versioned_data_kind import FEATURES, SEGMENTS pp = None @@ -37,9 +47,25 @@ def teardown_function(): pp.stop() -def setup_processor(config): +ONE_HOUR = 60 * 60 + + +def fast_retry_state(delay=0.05): + """A retry state whose every delay is ``delay``: small enough to skip the + real extended-regime wait, or large enough to prove a stop interrupts one.""" + return RetryState( + normal_initial_delay=delay, + normal_ceiling_delay=delay, + extended_initial_delay=delay, + extended_ceiling_delay=delay, + reset_policy=AfterConsecutiveSuccesses(POLLING_RESET_SUCCESSES), + operating_cadence=delay, + ) + + +def setup_processor(config, retry_state=None): global pp - pp = PollingUpdateProcessor(config, mock_requester, store, ready) + pp = PollingUpdateProcessor(config, mock_requester, store, ready, retry_state=retry_state) pp.start() @@ -77,12 +103,16 @@ def test_general_connection_error_does_not_cause_immediate_failure(ignore_mock): assert mock_requester.request_count >= 2 -def test_http_401_error_causes_immediate_failure(): - verify_unrecoverable_http_error(401) +def test_http_401_error_does_not_stop_polling(): + verify_unexpected_http_error(401) + + +def test_http_403_error_does_not_stop_polling(): + verify_unexpected_http_error(403) -def test_http_403_error_causes_immediate_failure(): - verify_unrecoverable_http_error(401) +def test_http_404_error_does_not_stop_polling(): + verify_unexpected_http_error(404) def test_http_408_error_does_not_cause_immediate_failure(): @@ -102,7 +132,10 @@ def test_http_503_error_does_not_cause_immediate_failure(): @mock.patch('ldclient.config.Config.poll_interval', new_callable=mock.PropertyMock, return_value=0.1) -def verify_unrecoverable_http_error(http_status_code, ignore_mock): +def verify_unexpected_http_error(http_status_code, ignore_mock): + """An error that needs a person to fix it -- a rejected SDK key, say -- is + still retried. It must not stop the poller, must not report OFF, and must + not falsely unblock initialization.""" spy = SpyListener() listeners = Listeners() listeners.add(spy) @@ -111,16 +144,275 @@ def verify_unrecoverable_http_error(http_status_code, ignore_mock): config._data_source_update_sink = DataSourceUpdateSinkImpl(store, listeners, Listeners()) mock_requester.exception = UnsuccessfulResponseException(http_status_code) - setup_processor(config) + setup_processor(config, retry_state=fast_retry_state()) finished = ready.wait(0.5) - assert finished + assert not finished assert not pp.initialized() + assert mock_requester.request_count >= 2 + + assert len(spy.statuses) > 1 + for status in spy.statuses: + assert status.state == DataSourceState.INITIALIZING + assert status.error.kind == DataSourceErrorKind.ERROR_RESPONSE + assert status.error.status_code == http_status_code + + +@mock.patch('ldclient.config.Config.poll_interval', new_callable=mock.PropertyMock, return_value=0.1) +def test_unexpected_http_error_moves_to_the_extended_regime(ignore_mock): + retry = for_polling(0.1) + mock_requester.exception = UnsuccessfulResponseException(401) + setup_processor(Config("SDK_KEY"), retry_state=retry) + + # The extended regime starts at five minutes, so only the first poll runs. + wait_until(lambda: retry.next_delay > 0.1) + assert not ready.wait(0.1) assert mock_requester.request_count == 1 - assert len(spy.statuses) == 1 - assert spy.statuses[0].state == DataSourceState.OFF - assert spy.statuses[0].error.kind == DataSourceErrorKind.ERROR_RESPONSE - assert spy.statuses[0].error.status_code == http_status_code + +def test_the_first_success_after_an_outage_polls_at_the_cadence(): + # A backoff wait applies to a retry, not to every operation. The retry + # state carries the wait, so this reads it there rather than measuring + # elapsed time. + with no_retry_jitter(): + retry = for_polling(30) + processor = PollingUpdateProcessor(Config("SDK_KEY"), mock_requester, store, ready, retry_state=retry) + + mock_requester.exception = UnsuccessfulResponseException(401) + processor._poll() + assert retry.next_delay == 5 * 60 + + mock_requester.exception = None + mock_requester.all_data = {FEATURES: {}, SEGMENTS: {}} + processor._poll() + assert retry.next_delay == 30 + assert retry._extended, "one success restores the cadence but does not reset" + + processor._poll() + assert retry.next_delay == 30 + assert not retry._extended, "two successes in a row reset the state" + + +@pytest.mark.parametrize( + "error", + [ + ssl.SSLCertVerificationError("unable to get local issuer certificate"), + ssl.SSLEOFError("EOF occurred in violation of protocol"), + ConnectionResetError(104, "reset by peer"), + ], + ids=["certificate", "peer-close-handshake", "reset"], +) +def test_transport_failures_poll_again_at_the_cadence(error): + """No transport failure reaches the extended regime, a bad certificate + included. Only an HTTP status can do that.""" + retry = for_polling(30) + processor = PollingUpdateProcessor(Config("SDK_KEY"), mock_requester, store, ready, retry_state=retry) + + mock_requester.exception = error + processor._poll() + assert retry.next_delay == 30 + assert not retry._extended + + +@mock.patch('ldclient.config.Config.poll_interval', new_callable=mock.PropertyMock, return_value=0.05) +def test_failure_transitions_from_valid(ignore_mock): + """A rejected SDK key after a poll has succeeded reports INTERRUPTED. OFF + is reserved for an explicit shutdown.""" + spy = SpyListener() + listeners = Listeners() + listeners.add(spy) + + config = Config("SDK_KEY") + config._data_source_update_sink = DataSourceUpdateSinkImpl(store, listeners, Listeners()) + + mock_requester.all_data = {FEATURES: {}, SEGMENTS: {}} + setup_processor(config, retry_state=fast_retry_state()) + assert ready.wait(2) + assert spy.statuses[0].state == DataSourceState.VALID + + mock_requester.exception = UnsuccessfulResponseException(401) + deadline = time.time() + 2 + while spy.statuses[-1].state == DataSourceState.VALID and time.time() < deadline: + time.sleep(0.01) + + assert spy.statuses[-1].state == DataSourceState.INTERRUPTED + assert spy.statuses[-1].error.kind == DataSourceErrorKind.ERROR_RESPONSE + assert spy.statuses[-1].error.status_code == 401 + assert all(s.state != DataSourceState.OFF for s in spy.statuses) + + +def test_second_start_is_a_no_op(): + """A second start() must not raise. Thread.start() would, so the processor + guards it.""" + mock_requester.all_data = {FEATURES: {}, SEGMENTS: {}} + setup_processor(Config("SDK_KEY")) + pp.start() + + assert ready.wait(2) + assert pp.initialized() + + +def test_the_log_reports_the_growing_retry_delay(caplog): + """The message has to carry the real delay, so someone reading logs can see + the backoff working.""" + caplog.set_level(logging.WARNING) + + with no_retry_jitter(): + retry = for_polling(30) + processor = PollingUpdateProcessor(Config("SDK_KEY"), mock_requester, store, ready, retry_state=retry) + mock_requester.exception = UnsuccessfulResponseException(401) + processor._poll() + processor._poll() + + messages = [r.getMessage() for r in caplog.records] + assert messages == [ + "Received HTTP error 401 (invalid SDK key) for polling request - will retry in 300.0s", + "Received HTTP error 401 (invalid SDK key) for polling request - will retry in 600.0s", + ] + # An error a person has to fix is logged at error level, every time. + assert [r.levelno for r in caplog.records] == [logging.ERROR, logging.ERROR] + + +def test_a_normal_failure_logs_the_poll_interval_at_warning_level(caplog): + caplog.set_level(logging.WARNING) + + with no_retry_jitter(): + retry = for_polling(30) + processor = PollingUpdateProcessor(Config("SDK_KEY"), mock_requester, store, ready, retry_state=retry) + mock_requester.exception = UnsuccessfulResponseException(503) + processor._poll() + + record = caplog.records[0] + assert record.getMessage() == "Received HTTP error 503 for polling request - will retry in 30.0s" + assert record.levelno == logging.WARNING + + +def test_a_transport_error_reports_a_delay_and_keeps_its_stacktrace(caplog): + caplog.set_level(logging.WARNING) + + with no_retry_jitter(): + retry = for_polling(30) + processor = PollingUpdateProcessor(Config("SDK_KEY"), mock_requester, store, ready, retry_state=retry) + mock_requester.exception = ConnectionResetError(104, "reset by peer") + processor._poll() + + record = caplog.records[0] + assert record.getMessage() == "Error encountered when updating flags: [Errno 104] reset by peer - will retry in 30.0s" + # The handler has exited by the time this is logged, so the exception has to + # be carried explicitly for the traceback to survive. + assert record.exc_info is not None + + +def _polling_thread(): + """Finds the task's worker thread by name, so a test can prove it exited + without reaching into the task's private state.""" + return next((t for t in threading.enumerate() if t.name == "ldclient.datasource.polling.repeating"), None) + + +def test_an_extended_regime_wait_is_cut_short_by_stop(): + """The reason the wait has to be interruptible at all. A 401 puts the next + poll five minutes out, and shutdown must not sit through it.""" + mock_requester.exception = UnsuccessfulResponseException(401) + retry = for_polling(30) + setup_processor(Config("SDK_KEY"), retry_state=retry) + + # Let the first poll happen, so the task is inside the long wait. + deadline = time.time() + 2 + while mock_requester.request_count < 1 and time.time() < deadline: + time.sleep(0.01) + assert mock_requester.request_count == 1 + assert retry._extended, "the wait under test should be minutes long" + + worker = _polling_thread() + assert worker is not None + + started = time.time() + pp.stop() + worker.join(2) + elapsed = time.time() - started + + # Without an interruptible wait this join would time out and the thread + # would still be sitting in a 300-second sleep. + assert not worker.is_alive() + assert elapsed < 1 + + +def test_an_absurd_poll_interval_does_not_kill_the_worker_thread(): + """``Event.wait`` raises above ``threading.TIMEOUT_MAX``, and that raise is + outside the try block around the poll, so the thread used to die while the + SDK still reported itself healthy.""" + mock_requester.all_data = {FEATURES: {}, SEGMENTS: {}} + setup_processor(Config("SDK_KEY", poll_interval=1e10)) + + assert ready.wait(2) + worker = _polling_thread() + assert worker is not None + + # A thread that raised on the wait exits as soon as the first poll returns. + worker.join(0.3) + assert worker.is_alive() + + +def test_stop_twice_and_stop_before_start_are_safe(): + """Neither a stop before the first poll nor a second stop should raise, + including while an hour-long wait is pending.""" + mock_requester.exception = UnsuccessfulResponseException(401) + processor = PollingUpdateProcessor( + Config("SDK_KEY"), mock_requester, store, ready, retry_state=fast_retry_state(ONE_HOUR) + ) + + processor.stop() + processor.stop() + processor.start() + processor.stop() + + +def test_stop_reports_off(): + spy = SpyListener() + listeners = Listeners() + listeners.add(spy) + + config = Config("SDK_KEY") + config._data_source_update_sink = DataSourceUpdateSinkImpl(store, listeners, Listeners()) + mock_requester.all_data = {FEATURES: {}, SEGMENTS: {}} + setup_processor(config) + assert ready.wait(2) + + pp.stop() + + assert spy.statuses[-1].state == DataSourceState.OFF + + +def test_a_poll_finishing_after_stop_reports_nothing(): + """The poll still in flight when stop() ran must not report after OFF.""" + spy = SpyListener() + listeners = Listeners() + listeners.add(spy) + + config = Config("SDK_KEY") + config._data_source_update_sink = DataSourceUpdateSinkImpl(store, listeners, Listeners()) + mock_requester.all_data = {FEATURES: {}, SEGMENTS: {}} + processor = PollingUpdateProcessor(config, mock_requester, store, ready) + + processor.stop() + processor._poll() + + assert [status.state for status in spy.statuses] == [DataSourceState.OFF] + + +def test_valid_status_is_reported_before_ready_is_set(): + # Mirrors go-server-sdk#442: a caller that wakes on readiness must not + # still be able to read INITIALIZING. + observed = [] + listeners = Listeners() + listeners.add(lambda status: observed.append((status.state, ready.is_set()))) + + config = Config("SDK_KEY") + config._data_source_update_sink = DataSourceUpdateSinkImpl(store, listeners, Listeners()) + mock_requester.all_data = {FEATURES: {}, SEGMENTS: {}} + setup_processor(config) + assert ready.wait(2) + + assert observed[0] == (DataSourceState.VALID, False) @mock.patch('ldclient.config.Config.poll_interval', new_callable=mock.PropertyMock, return_value=0.1) diff --git a/ldclient/testing/impl/datasource/test_streaming.py b/ldclient/testing/impl/datasource/test_streaming.py index 98c9d02a..307af278 100644 --- a/ldclient/testing/impl/datasource/test_streaming.py +++ b/ldclient/testing/impl/datasource/test_streaming.py @@ -1,15 +1,33 @@ +import logging +import ssl import time -from threading import Event +from threading import Event, Thread from typing import List import pytest +from ld_eventsource import SSEClient +from ld_eventsource.actions import Fault +from ld_eventsource.config import ( + ConnectStrategy, + ErrorStrategy, + RetryDelayStrategy +) +from ld_eventsource.errors import HTTPStatusError from ldclient.config import Config from ldclient.feature_store import InMemoryFeatureStore +from ldclient.impl.datasource.datasource_common import StreamClosedError from ldclient.impl.datasource.status import DataSourceUpdateSinkImpl from ldclient.impl.datasource.streaming import StreamingUpdateProcessor from ldclient.impl.events.diagnostics import _DiagnosticAccumulator from ldclient.impl.listeners import Listeners +from ldclient.impl.retry import ( + NORMAL_STREAMING_CEILING_DELAY, + STREAMING_RESET_INTERVAL, + AfterHealthyFor, + RetryState, + for_streaming +) from ldclient.interfaces import ( DataSourceErrorKind, DataSourceState, @@ -30,11 +48,31 @@ make_put_event, stream_content ) -from ldclient.testing.test_util import SpyListener +from ldclient.testing.sync_util import wait_until +from ldclient.testing.test_util import ( + SpyListener, + no_retry_jitter, + record_healthy_windows, + ticking_clock +) from ldclient.version import VERSION from ldclient.versioned_data_kind import FEATURES, SEGMENTS brief_delay = 0.001 +ONE_HOUR = 60 * 60 + + +def fast_retry_state(delay=brief_delay): + """A retry state whose every delay is ``delay``: small enough to skip the + real extended-regime wait, or large enough to prove a stop interrupts one.""" + return RetryState( + normal_initial_delay=delay, + normal_ceiling_delay=delay, + extended_initial_delay=delay, + extended_ceiling_delay=delay, + reset_policy=AfterHealthyFor(STREAMING_RESET_INTERVAL), + ) + # These long timeouts are necessary because of a problem in the Windows CI environment where HTTP requests to # the test server running at localhost tests are *extremely* slow. It looks like a similar issue to what's @@ -257,7 +295,9 @@ def test_recoverable_http_error(status): @pytest.mark.parametrize("status", [401, 403, 404]) -def test_unrecoverable_http_error(status): +def test_unexpected_http_error_backs_off_a_long_way(status): + """An error that needs a person to fix it does not stop the stream, but the + next attempt is five minutes out, so only one request is made here.""" error_handler = BasicResponse(status) store = InMemoryFeatureStore() ready = Event() @@ -269,11 +309,466 @@ def test_unrecoverable_http_error(status): with StreamingUpdateProcessor(config, store, ready, None) as sp: sp.start() - ready.wait(5) + server.wait_until_request_received() + # The next attempt is past the normal ceiling, so the failure + # has been recorded and the extended regime is in use. + wait_until(lambda: sp._retry.next_delay > NORMAL_STREAMING_CEILING_DELAY) + + # Initialization is not falsely unblocked. + assert not ready.wait(0.1) assert not sp.initialized() + assert sp.is_alive() server.should_have_requests(1) +@pytest.mark.parametrize("status", [401, 403, 404]) +def test_unexpected_http_error_keeps_retrying(status): + """The same failure with the delay compressed: the stream recovers once the + service does, rather than staying down for ever.""" + error_handler = BasicResponse(status) + store = InMemoryFeatureStore() + ready = Event() + with start_server() as server: + with stream_content(make_put_event()) as stream: + error_then_success = SequentialHandler(error_handler, stream) + config = Config(sdk_key='sdk-key', stream_uri=server.uri, initial_reconnect_delay=brief_delay) + + spy = SpyListener() + listeners = Listeners() + listeners.add(spy) + config._data_source_update_sink = DataSourceUpdateSinkImpl(store, listeners, Listeners()) + server.for_path('/all', error_then_success) + + with StreamingUpdateProcessor(config, store, ready, None, retry_state=fast_retry_state()) as sp: + sp.start() + ready.wait(start_wait) + assert sp.initialized() + server.should_have_requests(2) + + assert all(s.state != DataSourceState.OFF for s in spy.statuses) + assert spy.statuses[0].state == DataSourceState.INITIALIZING + assert spy.statuses[0].error.status_code == status + assert spy.statuses[-1].state == DataSourceState.VALID + + +def test_sse_client_hands_us_the_fault_before_it_waits(): + """Pins the ld_eventsource ordering the SDK relies on. + + The SDK computes and takes the retry delay itself, which only works + because SSEClient yields the Fault to the caller before its next connect + attempt sleeps. A library change that slept first would make this test + time out rather than fail quietly. + """ + with start_server() as server: + server.for_path('/all', BasicResponse(503)) + client = SSEClient( + connect=ConnectStrategy.http(url=server.uri + '/all'), + error_strategy=ErrorStrategy.always_continue(), + initial_retry_delay=30, + retry_delay_strategy=RetryDelayStrategy.default(max_delay=30, backoff_multiplier=2), + retry_delay_reset_threshold=0, + ) + try: + started = time.time() + first = next(iter(client.all)) + elapsed = time.time() - started + finally: + client.close() + + assert isinstance(first, Fault) + assert isinstance(first.error, HTTPStatusError) + # The library has a long delay queued up but has not taken it yet. + assert client.next_retry_delay >= 15 + assert elapsed < 5 + + +def test_the_sdk_configures_the_sse_client_never_to_wait(): + """The SDK owns the delay, so the library's own delay must stay at zero + however long the outage lasts.""" + store = InMemoryFeatureStore() + with start_server() as server: + server.for_path('/all', BasicResponse(503)) + config = Config(sdk_key='sdk-key', stream_uri=server.uri, initial_reconnect_delay=30) + sp = StreamingUpdateProcessor(config, store, Event(), None) + client = sp._create_sse_client() + try: + actions = iter(client.all) + first = next(actions) + second = next(actions) + finally: + client.close() + + assert isinstance(first, Fault) + assert isinstance(second, Fault) + assert client.next_retry_delay == 0 + + +def test_server_close_backs_off_and_keeps_the_stream_running(): + """The service normally leaves the connection open, so a clean close is a + connection failure: the SDK reports it and backs off, rather than + reconnecting in a tight loop.""" + store = InMemoryFeatureStore() + ready = Event() + flagv1 = FlagBuilder('flagkey').version(1).build() + flagv2 = FlagBuilder('flagkey').version(2).build() + + with start_server() as server: + with stream_content(make_put_event([flagv1])) as stream1: + with stream_content(make_put_event([flagv2])) as stream2: + config = Config(sdk_key='sdk-key', stream_uri=server.uri, initial_reconnect_delay=brief_delay) + + spy = SpyListener() + listeners = Listeners() + listeners.add(spy) + config._data_source_update_sink = DataSourceUpdateSinkImpl(store, listeners, Listeners()) + server.for_path('/all', SequentialHandler(stream1, stream2)) + + retry = fast_retry_state() + with StreamingUpdateProcessor(config, store, ready, None, retry_state=retry) as sp: + sp.start() + ready.wait(start_wait) + assert sp.initialized() + + stream1.close() + expect_update(store, FEATURES, flagv2) + + assert retry._attempts >= 1 + assert not retry._extended + + interrupted = [s for s in spy.statuses if s.state == DataSourceState.INTERRUPTED] + assert len(interrupted) >= 1 + assert interrupted[0].error.kind == DataSourceErrorKind.NETWORK_ERROR + + +def test_server_close_uses_the_normal_delay_curve(): + """A clean close is a NORMAL failure. Classifying it UNEXPECTED would put + a routine load-balancer drain into the extended regime and take a fleet + out of service for up to an hour.""" + store = InMemoryFeatureStore() + config = Config(sdk_key='sdk-key', initial_reconnect_delay=1) + retry = for_streaming(1) + sp = StreamingUpdateProcessor(config, store, Event(), None, retry_state=retry) + sp._running = True + sp._stop_event.set() # so the wait returns at once + + delays = [] + for _ in range(8): + sp._handle_error(StreamClosedError()) + delays.append(retry._max_delay) + + assert not retry._extended + assert delays == [30] * 8 + + +def test_our_own_interrupt_is_not_counted_as_a_server_close(): + """Bad JSON makes the SDK drop the connection itself. The SSE client then + reports that close as a Fault with no error, and counting it would record + the same failure twice and wait twice.""" + store = InMemoryFeatureStore() + ready = Event() + + with start_server() as server: + with stream_content(make_put_event()) as valid_stream, stream_content(make_invalid_put_event()) as invalid_stream: + config = Config(sdk_key='sdk-key', stream_uri=server.uri, initial_reconnect_delay=brief_delay) + + statuses: List[DataSourceStatus] = [] + listeners = Listeners() + + # The stream fixture holds the connection open, so it has to be + # closed for the server to move on to the next handler. This + # mirrors test_invalid_json_triggers_listener. + def listener(s): + if len(statuses) == 0: + invalid_stream.close() + statuses.append(s) + + listeners.add(listener) + + config._data_source_update_sink = DataSourceUpdateSinkImpl(store, listeners, Listeners()) + server.for_path('/all', SequentialHandler(invalid_stream, valid_stream)) + + retry = fast_retry_state() + with StreamingUpdateProcessor(config, store, ready, None, retry_state=retry) as sp: + sp.start() + ready.wait(start_wait) + assert sp.initialized() + server.should_have_requests(2) + + # One failure for the bad JSON, not a second for the close it + # caused. + assert retry._attempts == 1 + + +def test_a_leaked_interrupt_flag_does_not_swallow_a_server_close(): + """interrupt() is a no-op when the connection has already gone, so no + Fault arrives to clear the flag. A new connection must clear it, or the + next genuine close is recorded as ours and the backoff is skipped.""" + store = InMemoryFeatureStore() + ready = Event() + flagv1 = FlagBuilder('flagkey').version(1).build() + flagv2 = FlagBuilder('flagkey').version(2).build() + + with start_server() as server: + with stream_content(make_put_event([flagv1])) as stream1: + with stream_content(make_put_event([flagv2])) as stream2: + config = Config(sdk_key='sdk-key', stream_uri=server.uri, initial_reconnect_delay=brief_delay) + server.for_path('/all', SequentialHandler(stream1, stream2)) + + retry = fast_retry_state() + with StreamingUpdateProcessor(config, store, ready, None, retry_state=retry) as sp: + sp._interrupted_by_sdk = True + sp.start() + ready.wait(start_wait) + assert sp.initialized() + + stream1.close() + expect_update(store, FEATURES, flagv2) + + assert retry._attempts == 1 + + +def test_stop_before_start_does_not_raise(): + """stop() can land before run() has built the SSE client.""" + config = Config(sdk_key='sdk-key', stream_uri='http://localhost') + sp = StreamingUpdateProcessor(config, InMemoryFeatureStore(), Event(), None) + sp.stop() + + +def test_a_stop_before_the_connection_exists_still_ends_the_run(): + """stop() has nothing to close when run() has not built the client yet, so + the run itself must not go on to read a connection nobody will close.""" + store = InMemoryFeatureStore() + ready = Event() + + with start_server() as server: + with stream_content(make_put_event()) as stream: + server.for_path('/all', stream) + config = Config(sdk_key='sdk-key', stream_uri=server.uri) + sp = StreamingUpdateProcessor(config, store, ready, None) + + sp.stop() + thread = Thread(target=sp.run, daemon=True) + thread.start() + thread.join(update_wait) + + assert not thread.is_alive() + assert not ready.is_set() + assert not store.initialized + + +def test_a_raise_inside_the_loop_closes_the_connection(): + """A raise the loop does not catch must not leak the connection pool.""" + store = InMemoryFeatureStore() + closes: List[int] = [] + + with start_server() as server: + with stream_content(make_put_event()) as stream: + server.for_path('/all', stream) + config = Config(sdk_key='sdk-key', stream_uri=server.uri) + sp = StreamingUpdateProcessor(config, store, Event(), None) + + # KeyboardInterrupt is not an Exception, so the loop cannot catch it. + def explode(*args, **kwargs): + raise KeyboardInterrupt() + + sp._process_message = explode # type: ignore[method-assign] + + real_create = sp._create_sse_client + + def create_and_watch_close(): + client = real_create() + real_close = client.close + + def close(): + closes.append(1) + real_close() + + client.close = close # type: ignore[method-assign] + return client + + sp._create_sse_client = create_and_watch_close # type: ignore[method-assign] + + with pytest.raises(KeyboardInterrupt): + sp.run() + + assert closes == [1] + + +def _handle_errors_without_waiting(retry, errors): + """Drives _handle_error for each error and returns nothing. The stop event + is pre-set so the interruptible wait returns at once.""" + store = InMemoryFeatureStore() + config = Config(sdk_key='sdk-key', initial_reconnect_delay=1) + sp = StreamingUpdateProcessor(config, store, Event(), None, retry_state=retry) + sp._running = True + sp._stop_event.set() + for error in errors: + sp._handle_error(error) + + +def test_the_log_reports_the_growing_retry_delay(caplog): + """The message has to carry the real delay, so someone reading logs can see + the backoff working. The vaguer wording it replaced could not show this.""" + caplog.set_level(logging.WARNING) + + with no_retry_jitter(): + retry = for_streaming(1) + _handle_errors_without_waiting(retry, [HTTPStatusError(401), HTTPStatusError(401)]) + + messages = [r.getMessage() for r in caplog.records] + assert messages == [ + "Received HTTP error 401 (invalid SDK key) for stream connection - will retry in 300.0s", + "Received HTTP error 401 (invalid SDK key) for stream connection - will retry in 600.0s", + ] + # An error a person has to fix is logged at error level, every time. + assert [r.levelno for r in caplog.records] == [logging.ERROR, logging.ERROR] + + +def test_a_normal_failure_logs_a_short_delay_at_warning_level(caplog): + caplog.set_level(logging.WARNING) + + with no_retry_jitter(): + retry = for_streaming(1) + _handle_errors_without_waiting(retry, [HTTPStatusError(503), HTTPStatusError(503)]) + + messages = [r.getMessage() for r in caplog.records] + assert messages == [ + "Received HTTP error 503 for stream connection - will retry in 1.0s", + "Received HTTP error 503 for stream connection - will retry in 2.0s", + ] + assert [r.levelno for r in caplog.records] == [logging.WARNING, logging.WARNING] + + +def test_a_server_close_and_a_transport_error_both_report_a_delay(caplog): + caplog.set_level(logging.WARNING) + + with no_retry_jitter(): + retry = for_streaming(1) + _handle_errors_without_waiting(retry, [StreamClosedError(), ConnectionResetError(104, "reset by peer")]) + + messages = [r.getMessage() for r in caplog.records] + assert messages[0] == "The server closed the stream connection - will retry in 1.0s" + assert messages[1] == "Error on stream connection: [Errno 104] reset by peer - will retry in 2.0s" + + +def test_an_extended_regime_wait_is_cut_short_by_stop(): + """The reason the wait has to be interruptible at all. Shutdown must not + sit through an hour-long backoff.""" + store = InMemoryFeatureStore() + with start_server() as server: + config = Config(sdk_key='sdk-key', stream_uri=server.uri) + server.for_path('/all', BasicResponse(401)) + retry = fast_retry_state(ONE_HOUR) + + with StreamingUpdateProcessor(config, store, Event(), None, retry_state=retry) as sp: + sp.start() + server.wait_until_request_received() + # Confirm the wait under test really is long before measuring the stop. + wait_until(lambda: retry.next_delay > 60) + + started = time.time() + sp.stop() + sp.join(5) + elapsed = time.time() - started + + assert not sp.is_alive() + assert elapsed < 2, "stop() took %.2fs" % elapsed + + +def test_several_messages_on_one_stream_do_not_extend_the_reset_window(): + """The window starts at the first message and stays there, however many + more arrive on the same stream.""" + store = InMemoryFeatureStore() + ready = Event() + flag = FlagBuilder('flagkey').version(1).build() + + with start_server() as server: + with stream_content(make_put_event([flag]) + make_patch_event(FEATURES, flag)) as stream: + config = Config(sdk_key='sdk-key', stream_uri=server.uri, initial_reconnect_delay=brief_delay) + server.for_path('/all', stream) + + policy = AfterHealthyFor(STREAMING_RESET_INTERVAL) + retry = RetryState( + normal_initial_delay=brief_delay, + normal_ceiling_delay=brief_delay, + extended_initial_delay=brief_delay, + extended_ceiling_delay=brief_delay, + reset_policy=policy, + ) + # The clock moves on every read, so a window that had been + # restarted reads back as a different time. + windows = record_healthy_windows(policy) + with ticking_clock(): + with StreamingUpdateProcessor(config, store, ready, None, retry_state=retry) as sp: + sp.start() + ready.wait(start_wait) + assert sp.initialized() + expect_update(store, FEATURES, flag) + + assert len(windows) >= 2, "both messages should have signalled" + assert len(set(windows)) == 1, "the window moved between messages" + + +def test_a_fresh_stream_starts_a_new_reset_window(): + """A stream teardown clears the window through record_failure, so the next + stream measures its own stretch rather than inheriting the old one.""" + store = InMemoryFeatureStore() + ready = Event() + flagv1 = FlagBuilder('flagkey').version(1).build() + flagv2 = FlagBuilder('flagkey').version(2).build() + + with start_server() as server: + with stream_content(make_put_event([flagv1])) as stream1: + with stream_content(make_put_event([flagv2])) as stream2: + config = Config(sdk_key='sdk-key', stream_uri=server.uri, initial_reconnect_delay=brief_delay) + server.for_path('/all', SequentialHandler(stream1, stream2)) + + policy = AfterHealthyFor(STREAMING_RESET_INTERVAL) + retry = RetryState( + normal_initial_delay=brief_delay, + normal_ceiling_delay=brief_delay, + extended_initial_delay=brief_delay, + extended_ceiling_delay=brief_delay, + reset_policy=policy, + ) + windows = record_healthy_windows(policy) + with ticking_clock(): + with StreamingUpdateProcessor(config, store, ready, None, retry_state=retry) as sp: + sp.start() + ready.wait(start_wait) + assert sp.initialized() + + stream1.close() + expect_update(store, FEATURES, flagv2) + + assert len(set(windows)) == 2, "the second stream reused the first window" + + +@pytest.mark.parametrize( + "error", + [ + ssl.SSLCertVerificationError("unable to get local issuer certificate"), + ssl.SSLEOFError("EOF occurred in violation of protocol"), + ConnectionResetError(104, "reset by peer"), + ], + ids=["certificate", "peer-close-handshake", "reset"], +) +def test_transport_failures_stay_in_the_normal_regime(error): + """No transport failure reaches the extended regime, a bad certificate + included. Only an HTTP status can do that.""" + store = InMemoryFeatureStore() + config = Config(sdk_key='sdk-key', initial_reconnect_delay=1) + retry = for_streaming(1) + sp = StreamingUpdateProcessor(config, store, Event(), None, retry_state=retry) + sp._running = True + sp._stop_event.set() # so the wait returns at once + + sp._handle_error(error) + + assert not retry._extended + assert retry._max_delay == 30 + + def test_http_proxy(monkeypatch): def _stream_processor_proxy_test(server, config, secure): store = InMemoryFeatureStore() @@ -407,6 +902,8 @@ def listener(s): def test_failure_transitions_from_valid(): + """A rejected SDK key after the stream was valid reports INTERRUPTED. OFF + is reserved for an explicit shutdown.""" store = InMemoryFeatureStore() ready = Event() error_handler = BasicResponse(401) @@ -426,14 +923,18 @@ def test_failure_transitions_from_valid(): with StreamingUpdateProcessor(config, store, ready, None) as sp: sp.start() - ready.wait(start_wait) + server.wait_until_request_received() + wait_until(lambda: len(spy.statuses) == 2) + + # The 401 is retried five minutes out, so readiness never fires. + assert not ready.wait(0.1) server.should_have_requests(1) assert len(spy.statuses) == 2 assert spy.statuses[0].state == DataSourceState.VALID - assert spy.statuses[1].state == DataSourceState.OFF + assert spy.statuses[1].state == DataSourceState.INTERRUPTED assert spy.statuses[1].error.kind == DataSourceErrorKind.ERROR_RESPONSE assert spy.statuses[1].error.status_code == 401 diff --git a/ldclient/testing/impl/test_data_sink.py b/ldclient/testing/impl/test_data_sink.py index d905db78..1ec5d49e 100644 --- a/ldclient/testing/impl/test_data_sink.py +++ b/ldclient/testing/impl/test_data_sink.py @@ -1,3 +1,4 @@ +import time from typing import Callable, Dict import mock @@ -6,7 +7,11 @@ from ldclient.feature_store import InMemoryFeatureStore from ldclient.impl.datasource.status import DataSourceUpdateSinkImpl from ldclient.impl.listeners import Listeners -from ldclient.interfaces import DataSourceErrorKind, DataSourceState +from ldclient.interfaces import ( + DataSourceErrorInfo, + DataSourceErrorKind, + DataSourceState +) from ldclient.testing.builders import ( FlagBuilder, FlagRuleBuilder, @@ -82,6 +87,42 @@ def test_interrupting_initializing_stays_initializing(): assert sink.status.error is None +def test_off_is_terminal(): + spy = SpyListener() + status_listener = Listeners() + status_listener.add(spy) + + sink = DataSourceUpdateSinkImpl(InMemoryFeatureStore(), status_listener, Listeners()) + sink.update_status(DataSourceState.VALID, None) + sink.update_status(DataSourceState.OFF, None) + + # A poll or stream connection still in flight when the data source stopped. + # Test both a plain state change and one carrying an error, so neither can + # get through. + sink.update_status(DataSourceState.VALID, None) + sink.update_status(DataSourceState.INTERRUPTED, DataSourceErrorInfo(DataSourceErrorKind.NETWORK_ERROR, 0, time.time(), 'late')) + + assert sink.status.state == DataSourceState.OFF + assert sink.status.error is None + assert [status.state for status in spy.statuses] == [DataSourceState.VALID, DataSourceState.OFF] + + +@mock.patch('ldclient.feature_store.InMemoryFeatureStore.init', side_effect=[Exception('cannot init')]) +def test_store_error_after_off_reports_nothing(mock_init, prereq_data): + spy = SpyListener() + status_listener = Listeners() + status_listener.add(spy) + + sink = DataSourceUpdateSinkImpl(InMemoryFeatureStore(), status_listener, Listeners()) + sink.update_status(DataSourceState.OFF, None) + + with pytest.raises(Exception): + sink.init(prereq_data) + + assert sink.status.state == DataSourceState.OFF + assert [status.state for status in spy.statuses] == [DataSourceState.OFF] + + def test_listener_is_only_triggered_for_state_changes(): spy = SpyListener() status_listener = Listeners() diff --git a/ldclient/testing/impl/test_retry.py b/ldclient/testing/impl/test_retry.py index bb947037..138f5182 100644 --- a/ldclient/testing/impl/test_retry.py +++ b/ldclient/testing/impl/test_retry.py @@ -15,10 +15,12 @@ import pytest +from ldclient.config import ( + DEFAULT_INITIAL_RECONNECT_DELAY, + DEFAULT_POLL_INTERVAL +) from ldclient.impl import retry from ldclient.impl.retry import ( - DEFAULT_POLL_INTERVAL, - DEFAULT_STREAMING_INITIAL_RECONNECT_DELAY, EXTENDED_CEILING_DELAY, EXTENDED_INITIAL_DELAY, NORMAL_STREAMING_CEILING_DELAY, @@ -110,9 +112,9 @@ def test_non_error_statuses_are_normal(self, status): class TestFactoryInputGuards: - """``Config`` does not check ``initial_reconnect_delay`` at all, and only - clamps ``poll_interval``. A non-positive value would retry with no wait; a - non-finite one makes the jitter arithmetic produce NaN.""" + """``Config`` reports these options as configured, so the factories are the + only guard. A non-positive value would retry with no wait; a non-finite one + makes the jitter arithmetic produce NaN.""" @pytest.mark.parametrize( "configured", @@ -125,12 +127,11 @@ def test_streaming_falls_back_to_the_default(self, configured, caplog): state = for_streaming(configured) delay = failure_delay(state, NORMAL) - assert state._min_delay == DEFAULT_STREAMING_INITIAL_RECONNECT_DELAY - assert delay == DEFAULT_STREAMING_INITIAL_RECONNECT_DELAY + assert state._min_delay == DEFAULT_INITIAL_RECONNECT_DELAY + assert delay == DEFAULT_INITIAL_RECONNECT_DELAY assert math.isfinite(delay) and delay > 0 assert caplog.records[0].getMessage() == ( - "initial_reconnect_delay must be a positive, finite number of seconds; " - "using the default of 1s" + "initial_reconnect_delay must be a positive, finite number of seconds; using the default of 1s" ) @pytest.mark.parametrize( @@ -148,11 +149,10 @@ def test_polling_falls_back_to_the_default(self, configured, caplog): assert delay == DEFAULT_POLL_INTERVAL assert math.isfinite(delay) and delay > 0 assert caplog.records[0].getMessage() == ( - "poll_interval must be a positive, finite number of seconds; " - "using the default of 30s" + "poll_interval must be a positive, finite number of seconds; using the default of 30s" ) - @pytest.mark.parametrize("configured", [0.001, 0.5, 1, 5, 45]) + @pytest.mark.parametrize("configured", [0.001, 0.5, 1, 5, 30]) def test_a_positive_streaming_delay_is_left_alone(self, configured, caplog): caplog.set_level(logging.WARNING) @@ -173,6 +173,24 @@ def test_a_positive_poll_interval_is_left_alone(self, configured, caplog): assert caplog.records == [] +class TestStreamingCeilings: + """A configured reconnect delay longer than a regime's ceiling raises that + bound. The configured value wins over our default, rather than being cut + down to it.""" + + @pytest.mark.parametrize("configured", [600, 7200, 86400]) + def test_a_delay_past_the_normal_ceiling_raises_the_bound(self, configured): + assert failure_delay(for_streaming(configured), NORMAL) == configured + + @pytest.mark.parametrize("configured", [7200, 86400]) + def test_a_delay_past_the_extended_ceiling_raises_the_bound(self, configured): + assert failure_delay(for_streaming(configured), UNEXPECTED) == configured + + @pytest.mark.parametrize("configured", [0.5, 1, 30]) + def test_a_delay_within_the_normal_ceiling_is_untouched(self, configured): + assert failure_delay(for_streaming(configured), NORMAL) == configured + + class TestStreamingExtendedDelayFloor: """A delay that applies after an unexpected failure must not be below the component's initial delay.""" @@ -217,11 +235,11 @@ def test_extended_regime_doubles_up_to_the_ceiling(self): delays += [failure_delay(state, NORMAL) for _ in range(5)] assert delays == [5 * 60, 10 * 60, 20 * 60, 40 * 60, 60 * 60, 60 * 60] - def test_a_configured_initial_delay_raises_the_ceiling_with_it(self): - # The ceiling must not fall below the initial delay. - state = for_streaming(45) - assert state._max_delay == 45 - assert failure_delay(state, NORMAL) == 45 + @pytest.mark.parametrize("configured", [1, 30, 45, 600]) + def test_the_ceiling_is_never_below_what_was_configured(self, configured): + # A configured delay longer than the normal ceiling raises the bound, so + # the first wait is never shorter than what the caller asked for. + assert for_streaming(configured)._max_delay == max(NORMAL_STREAMING_CEILING_DELAY, configured) def test_the_ceiling_is_sticky_once_the_extended_regime_starts(self): # A normal failure after an unexpected one must not lower the bounds @@ -393,12 +411,20 @@ def test_the_wait_never_falls_below_the_poll_interval(self): assert failure_delay(state, UNEXPECTED) >= 30 def test_a_poll_interval_longer_than_the_extended_bounds_wins(self): - # The ceiling is lifted by record_failure clamping it against the - # initial delay, not by for_polling clamping the ceiling itself. - state = for_polling(2 * 60 * 60) - assert failure_delay(state, UNEXPECTED) == 2 * 60 * 60 - assert state._max_delay == 2 * 60 * 60 - assert state._min_delay == 2 * 60 * 60 + """Unlike the streaming delay, the poll interval is not clamped to the + extended ceiling. Nothing may poll faster than the configured interval, + so the cadence wins where the two conflict.""" + two_hours = 2 * 60 * 60 + assert two_hours > EXTENDED_CEILING_DELAY + + state = for_polling(two_hours) + + assert failure_delay(state, UNEXPECTED) == two_hours + assert state._max_delay == two_hours + assert state._min_delay == two_hours + assert failure_delay(state, NORMAL) == two_hours + state.record_success() + assert state.next_delay == two_hours def test_one_success_restores_the_cadence_while_the_state_is_still_raised(self): # A backoff wait applies to a retry, not to every operation. diff --git a/ldclient/testing/test_config.py b/ldclient/testing/test_config.py index 3b18c3cf..9595eb0d 100644 --- a/ldclient/testing/test_config.py +++ b/ldclient/testing/test_config.py @@ -1,6 +1,12 @@ import pytest -from ldclient.config import Config +from ldclient.async_config import AsyncConfig +from ldclient.config import DEFAULT_POLL_INTERVAL, Config + +# Both classes handle these options identically, so every case runs against +# both rather than being duplicated and left to drift. +CONFIG_CLASSES = [Config, AsyncConfig] +CONFIG_IDS = ["Config", "AsyncConfig"] def test_copy_config(): @@ -38,14 +44,32 @@ def test_with_wrapper_information_defaults_the_version(): assert wrapped.wrapper_version is None -def test_can_set_valid_poll_interval(): - config = Config(sdk_key="SDK_KEY", poll_interval=31) - assert config.poll_interval == 31 - - -def test_minimum_poll_interval_is_enforced(): - config = Config(sdk_key="SDK_KEY", poll_interval=29) - assert config.poll_interval == 30 +@pytest.mark.parametrize("config_class", CONFIG_CLASSES, ids=CONFIG_IDS) +@pytest.mark.parametrize( + "configured,expected", + [ + (5, DEFAULT_POLL_INTERVAL), + (29, DEFAULT_POLL_INTERVAL), + (30, 30), + (31, 31), + (60, 60), + ], + ids=["below-the-minimum", "just-below", "at-the-minimum", "just-above", "above"], +) +def test_a_poll_interval_below_the_minimum_is_raised_to_it(config_class, configured, expected): + config = config_class(sdk_key="SDK_KEY", poll_interval=configured) + + assert config.poll_interval == expected + + +@pytest.mark.parametrize("config_class", CONFIG_CLASSES, ids=CONFIG_IDS) +@pytest.mark.parametrize("configured", [0.001, 0.5, 1, 30, 600], ids=["tiny", "fraction", "default", "thirty", "long"]) +def test_a_configured_initial_reconnect_delay_is_reported_as_given(config_class, configured): + """This option has no minimum, so a sub-second value survives as given. + The spec ceilings are applied by ``for_streaming``, not here.""" + config = config_class(sdk_key="SDK_KEY", initial_reconnect_delay=configured) + + assert config.initial_reconnect_delay == configured def test_can_set_valid_diagnostic_interval(): diff --git a/ldclient/testing/test_ldclient_end_to_end.py b/ldclient/testing/test_ldclient_end_to_end.py index 8e608d14..f5b5b9b0 100644 --- a/ldclient/testing/test_ldclient_end_to_end.py +++ b/ldclient/testing/test_ldclient_end_to_end.py @@ -1,5 +1,6 @@ import json import sys +import time import pytest @@ -53,12 +54,22 @@ def test_client_starts_in_streaming_mode(): assert r.headers['Authorization'] == sdk_key -def test_client_fails_to_start_in_streaming_mode_with_401_error(): +def test_client_does_not_initialize_in_streaming_mode_with_401_error(): + """A rejected SDK key no longer fails fast. The constructor waits out the + full start_wait and returns uninitialized, while the SDK keeps retrying in + the background.""" with start_server() as stream_server: stream_server.for_path('/all', BasicResponse(401)) config = Config(sdk_key=sdk_key, stream_uri=stream_server.uri, send_events=False) - with LDClient(config=config) as client: + start_wait = 0.5 + started = time.monotonic() + with LDClient(config=config, start_wait=start_wait) as client: + elapsed = time.monotonic() - started + # A bound rather than the exact start_wait: Event.wait can return a + # fraction early against a separate clock. Failing fast took + # milliseconds, so this still catches it. + assert elapsed >= start_wait / 2 assert not client.is_initialized() assert client.variation(always_true_flag['key'], user, False) is False @@ -91,12 +102,18 @@ def test_client_starts_in_polling_mode(): assert r.headers['Authorization'] == sdk_key -def test_client_fails_to_start_in_polling_mode_with_401_error(): +def test_client_does_not_initialize_in_polling_mode_with_401_error(): + """As with streaming, a rejected SDK key no longer fails fast.""" with start_server() as poll_server: poll_server.for_path('/sdk/latest-all', BasicResponse(401)) config = Config(sdk_key=sdk_key, base_uri=poll_server.uri, stream=False, send_events=False) - with LDClient(config=config) as client: + start_wait = 0.5 + started = time.monotonic() + with LDClient(config=config, start_wait=start_wait) as client: + elapsed = time.monotonic() - started + # See the streaming case above: a bound, not the exact start_wait. + assert elapsed >= start_wait / 2 assert not client.is_initialized() assert client.variation(always_true_flag['key'], user, False) is False