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 031faf4d..edf9ef12 100644 --- a/ldclient/async_client.py +++ b/ldclient/async_client.py @@ -384,8 +384,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/client.py b/ldclient/client.py index 2329defa..3dae913c 100644 --- a/ldclient/client.py +++ b/ldclient/client.py @@ -311,10 +311,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/impl/aio/concurrency.py b/ldclient/impl/aio/concurrency.py index ed3f7962..c4b6ad70 100644 --- a/ldclient/impl/aio/concurrency.py +++ b/ldclient/impl/aio/concurrency.py @@ -9,11 +9,11 @@ import asyncio import inspect -import time from queue import Empty as QueueEmpty # noqa: F401 (shared timeout exception) from queue import Full as QueueFull # noqa: F401 (shared capacity exception) from typing import Any, Callable, Coroutine, Optional, Set +from ldclient.impl.repeating_task import DelaySource, FixedDelay from ldclient.impl.util import log @@ -189,25 +189,31 @@ async def stop_all(self, timeout: float = 1) -> None: class AsyncRepeatingTask: - """Calls a callback repeatedly at fixed intervals on a background task. + """Calls a callback repeatedly on a background task, waiting whatever its + :class:`~ldclient.impl.repeating_task.DelaySource` gives. Mirrors the semantics of ``ldclient.impl.repeating_task.RepeatingTask``: - the interval is measured from the start of each invocation, exceptions - from the callback are logged, and ``stop()`` prevents any further - invocations but cannot be undone.""" + the wait starts when the callback returns, exceptions from the callback + are logged, and ``stop()`` prevents any further invocations but cannot be + undone.""" - def __init__(self, label: str, interval: float, initial_delay: float, callable: Callable): + def __init__(self, label: str, delays: DelaySource, initial_delay: float, callable: Callable): self.__label = label - self.__interval = interval + self.__delays = delays self.__initial_delay = initial_delay self.__action = callable self.__stop = AsyncEvent() self.__task: Optional[asyncio.Task] = None + @staticmethod + def at_interval(label: str, interval: float, initial_delay: float, callable: Callable) -> 'AsyncRepeatingTask': + """Creates a task that runs at a fixed interval.""" + return AsyncRepeatingTask(label, FixedDelay(interval), initial_delay, callable) + def start(self): - """Starts the background task. Like a thread, the task can only be - started once.""" + """Starts the background task, if it is not running already.""" if self.__task is not None: - raise RuntimeError("tasks can only be started once") + log.info("Task %s has already been started; ignoring" % self.__label) + return self.__task = asyncio.ensure_future(self._run()) try: self.__task.set_name(f"{self.__label}.repeating") @@ -215,7 +221,9 @@ def start(self): pass def stop(self): - """Tells the background task to stop. It cannot be restarted after this.""" + """Tells the background task to stop. + + The stop is permanent. A later ``start()`` does not resume the task.""" self.__stop.set() task = self.__task # When stop() is called from within the action itself, let the loop @@ -237,14 +245,15 @@ async def _run(self): return stopped = self.__stop.is_set() while not stopped: - next_time = time.time() + self.__interval try: result = self.__action() if inspect.isawaitable(result): await result except Exception as e: log.exception("Unexpected exception on worker task: %s" % e) - delay = next_time - time.time() + # The wait starts when the callback returns, so a slow callback + # never shortens it. + delay = self.__delays.next_delay if delay > 0: stopped = await self.__stop.wait(delay) else: diff --git a/ldclient/impl/aio/transport.py b/ldclient/impl/aio/transport.py index e9aa374f..6226c5d2 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: + # A zero base delay plus the no-op base strategy holds + # next_retry_delay at zero, so the SSE client never sleeps. + 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/async_big_segments.py b/ldclient/impl/async_big_segments.py index ce4aecd0..16315e0a 100644 --- a/ldclient/impl/async_big_segments.py +++ b/ldclient/impl/async_big_segments.py @@ -65,7 +65,7 @@ def __init__(self, config: AsyncBigSegmentsConfig): if self.__store: self.__cache = ExpiringDict(max_len=config.context_cache_size, max_age_seconds=config.context_cache_time) - self.__poll_task = AsyncRepeatingTask("ldclient.bigsegment.status-poll", config.status_poll_interval, 0, self.poll_store_and_update_status) + self.__poll_task = AsyncRepeatingTask.at_interval("ldclient.bigsegment.status-poll", config.status_poll_interval, 0, self.poll_store_and_update_status) def start(self): """Starts the status polling task. Separated from __init__ so the manager diff --git a/ldclient/impl/big_segments.py b/ldclient/impl/big_segments.py index cf2dec61..2d5dcf43 100644 --- a/ldclient/impl/big_segments.py +++ b/ldclient/impl/big_segments.py @@ -67,7 +67,7 @@ def __init__(self, config: BigSegmentsConfig): if self.__store: self.__cache = ExpiringDict(max_len=config.context_cache_size, max_age_seconds=config.context_cache_time) - self.__poll_task = RepeatingTask("ldclient.bigsegment.status-poll", config.status_poll_interval, 0, self.poll_store_and_update_status) + self.__poll_task = RepeatingTask.at_interval("ldclient.bigsegment.status-poll", config.status_poll_interval, 0, self.poll_store_and_update_status) self.__poll_task.start() def stop(self): diff --git a/ldclient/impl/datasource/async_polling.py b/ldclient/impl/datasource/async_polling.py index d0530a7c..3a638ed8 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("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,7 +57,12 @@ def initialized(self): return self._ready.is_set() and self._store.initialized async def stop(self): - self.__stop_with_error_info(None) + log.info("Stopping AsyncPollingUpdateProcessor") + self._task.stop() + + if self._data_source_update_sink is not None: + self._data_source_update_sink.update_status(DataSourceState.OFF, 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. @@ -52,39 +71,41 @@ async def stop(self): 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 - - self._data_source_update_sink.update_status(DataSourceState.OFF, error) - - 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: + # A certificate failure lands here too, and is as normal as the rest. + 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) + delay = self._retry.record_failure(kind) + 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_streaming.py b/ldclient/impl/datasource/async_streaming.py index 002d7b65..f5afc506 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,7 +85,7 @@ 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): @@ -82,21 +95,25 @@ async def _run(self): 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 +126,17 @@ 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 means the connection closed cleanly. + # If we asked for that close, we have already recorded the + # failure behind it and must not record it twice. Otherwise + # the server closed a connection it normally leaves open, + # which is a connection failure the SDK backs off from. 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): @@ -141,12 +166,10 @@ def _record_stream_init(self, failed: bool): 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. + # loop could run _run at the teardown 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 if self._sse: @@ -156,7 +179,15 @@ async def __stop_with_error_info(self, error: Optional[DataSourceErrorInfo]): if self._data_source_update_sink is None: return - self._data_source_update_sink.update_status(DataSourceState.OFF, error) + # OFF means an explicit shutdown. No stream failure produces it. + self._data_source_update_sink.update_status(DataSourceState.OFF, None) + + 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 +229,53 @@ 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: + # A certificate failure lands here too, and is as normal as the rest. + 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) + delay = self._retry.record_failure(kind) + 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 + self._connection_attempt_start_time = time.time() + delay + if delay > 0: + await asyncio.sleep(delay) + 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 171df9eb..504cb7d8 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("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))) + # A certificate failure lands here too, and is as normal as the rest. + 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 + + delay = self._retry.record_failure(kind) + 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/streaming.py b/ldclient/impl/datasource/streaming.py index e5496147..935d3823 100644 --- a/ldclient/impl/datasource/streaming.py +++ b/ldclient/impl/datasource/streaming.py @@ -1,7 +1,8 @@ import json import time +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 +16,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 +41,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,7 +61,10 @@ 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._stop_event = ThreadEvent() + self._interrupted_by_sdk = False def run(self): log.info("Starting StreamingUpdateProcessor connecting to uri: " + self._uri) @@ -67,21 +76,25 @@ def run(self): 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._sse.interrupt() + self._interrupt_stream() - self._handle_error(e) + if not self._handle_error(e): + break except Exception as e: log.info("Error while handling stream event; will restart stream: %s" % e) - self._sse.interrupt() + self._interrupt_stream() - if self._data_source_update_sink is not None: - error_info = DataSourceErrorInfo(DataSourceErrorKind.UNKNOWN, 0, time.time(), str(e)) + if not 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) @@ -94,9 +107,17 @@ def run(self): 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. + # A Fault with no error means the connection closed cleanly. If + # we asked for that close, we have already recorded the failure + # behind it and must not record it twice. Otherwise the server + # closed a connection it normally leaves open, which is a + # connection failure the SDK backs off from. 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): @@ -118,25 +139,38 @@ 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 SDK owns the retry delay, so the SSE client must never wait. + # A zero base delay plus the no-op base strategy holds + # next_retry_delay at zero, which is what these three arguments + # are for. The SSE client hands us the Fault before it would + # sleep, so we classify the failure and wait ourselves in + # _handle_error. Our wait is interruptible, which matters because + # the extended regime can ask for an hour. + 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() 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) + # OFF means an explicit shutdown. No stream failure produces it. + self._data_source_update_sink.update_status(DataSourceState.OFF, None) + + 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 +210,49 @@ 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: + # A certificate failure lands here too, and is as normal as the rest. + 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) + delay = self._retry.record_failure(kind) + 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 + self._connection_attempt_start_time = time.time() + delay + return not self._stop_event.wait(delay) # magic methods for "with" statement (used in testing) def __enter__(self): diff --git a/ldclient/impl/datasystem/async_fdv2.py b/ldclient/impl/datasystem/async_fdv2.py index c9d0388d..612fe4cb 100644 --- a/ldclient/impl/datasystem/async_fdv2.py +++ b/ldclient/impl/datasystem/async_fdv2.py @@ -148,7 +148,7 @@ def _update_availability(self, available: bool) -> None: else: log.warning("Detected persistent store unavailability; updates will be cached until it recovers") if self._poller is None: - task_to_start = AsyncRepeatingTask("ldclient.check-availability", 0.5, 0, self._check_availability) + task_to_start = AsyncRepeatingTask.at_interval("ldclient.check-availability", 0.5, 0, self._check_availability) self._poller = task_to_start self._status_sink(DataStoreStatus(available, True)) @@ -545,7 +545,7 @@ async def _consume_synchronizer_results( :return: the ConditionDirective describing how to proceed """ action_queue: AsyncQueue = AsyncQueue() - timer = AsyncRepeatingTask( + timer = AsyncRepeatingTask.at_interval( label="AsyncFDv2-sync-cond-timer", interval=10, initial_delay=10, diff --git a/ldclient/impl/datasystem/fdv1.py b/ldclient/impl/datasystem/fdv1.py index 38655415..09ca6939 100644 --- a/ldclient/impl/datasystem/fdv1.py +++ b/ldclient/impl/datasystem/fdv1.py @@ -101,7 +101,7 @@ def __update_availability(self, available: bool): return log.warn("Detected persistent store unavailability; updates will be cached until it recovers") - task = RepeatingTask("ldclient.check-availability", 0.5, 0, self.__check_availability) + task = RepeatingTask.at_interval("ldclient.check-availability", 0.5, 0, self.__check_availability) with self.__lock.write(): self.__poller = task diff --git a/ldclient/impl/datasystem/fdv2.py b/ldclient/impl/datasystem/fdv2.py index 3455eab1..75f8d28c 100644 --- a/ldclient/impl/datasystem/fdv2.py +++ b/ldclient/impl/datasystem/fdv2.py @@ -129,7 +129,7 @@ def __update_availability(self, available: bool): poller_to_stop = self.__poller self.__poller = None elif self.__poller is None: - task_to_start = RepeatingTask("ldclient.check-availability", 0.5, 0, self.__check_availability) + task_to_start = RepeatingTask.at_interval("ldclient.check-availability", 0.5, 0, self.__check_availability) self.__poller = task_to_start if available: @@ -536,7 +536,7 @@ def _consume_synchronizer_results( :return: the ConditionDirective describing how to proceed """ action_queue: Queue = Queue() - timer = RepeatingTask( + timer = RepeatingTask.at_interval( label="FDv2-sync-cond-timer", interval=10, initial_delay=10, diff --git a/ldclient/impl/events/async_event_processor.py b/ldclient/impl/events/async_event_processor.py index 1e5b0215..dafd2483 100644 --- a/ldclient/impl/events/async_event_processor.py +++ b/ldclient/impl/events/async_event_processor.py @@ -202,13 +202,13 @@ class DefaultAsyncEventProcessor(AsyncEventProcessor): def __init__(self, config: AsyncConfig, http=None, dispatcher_class=None, diagnostic_accumulator=None): self._inbox = AsyncQueue(config.events_max_pending) self._inbox_full = False - self._flush_timer = AsyncRepeatingTask("ldclient.events.flush", config.flush_interval, config.flush_interval, self.flush) - self._contexts_flush_timer = AsyncRepeatingTask("ldclient.events.context-flush", config.context_keys_flush_interval, config.context_keys_flush_interval, self._flush_contexts) + self._flush_timer = AsyncRepeatingTask.at_interval("ldclient.events.flush", config.flush_interval, config.flush_interval, self.flush) + self._contexts_flush_timer = AsyncRepeatingTask.at_interval("ldclient.events.context-flush", config.context_keys_flush_interval, config.context_keys_flush_interval, self._flush_contexts) self._flush_timer.start() self._contexts_flush_timer.start() self._diagnostic_event_timer: Optional[AsyncRepeatingTask] if diagnostic_accumulator is not None: - self._diagnostic_event_timer = AsyncRepeatingTask("ldclient.events.send-diagnostic", config.diagnostic_recording_interval, config.diagnostic_recording_interval, self._send_diagnostic) + self._diagnostic_event_timer = AsyncRepeatingTask.at_interval("ldclient.events.send-diagnostic", config.diagnostic_recording_interval, config.diagnostic_recording_interval, self._send_diagnostic) self._diagnostic_event_timer.start() else: self._diagnostic_event_timer = None diff --git a/ldclient/impl/events/event_processor.py b/ldclient/impl/events/event_processor.py index 20cf03a4..6581070a 100644 --- a/ldclient/impl/events/event_processor.py +++ b/ldclient/impl/events/event_processor.py @@ -176,12 +176,12 @@ class DefaultEventProcessor(EventProcessor): def __init__(self, config, http=None, dispatcher_class=None, diagnostic_accumulator=None): self._inbox = queue.Queue(config.events_max_pending) self._inbox_full = False - self._flush_timer = RepeatingTask("ldclient.events.flush", config.flush_interval, config.flush_interval, self.flush) - self._contexts_flush_timer = RepeatingTask("ldclient.events.context-flush", config.context_keys_flush_interval, config.context_keys_flush_interval, self._flush_contexts) + self._flush_timer = RepeatingTask.at_interval("ldclient.events.flush", config.flush_interval, config.flush_interval, self.flush) + self._contexts_flush_timer = RepeatingTask.at_interval("ldclient.events.context-flush", config.context_keys_flush_interval, config.context_keys_flush_interval, self._flush_contexts) self._flush_timer.start() self._contexts_flush_timer.start() if diagnostic_accumulator is not None: - self._diagnostic_event_timer = RepeatingTask("ldclient.events.send-diagnostic", config.diagnostic_recording_interval, config.diagnostic_recording_interval, self._send_diagnostic) + self._diagnostic_event_timer = RepeatingTask.at_interval("ldclient.events.send-diagnostic", config.diagnostic_recording_interval, config.diagnostic_recording_interval, self._send_diagnostic) self._diagnostic_event_timer.start() else: self._diagnostic_event_timer = None diff --git a/ldclient/impl/integrations/files/file_data_source.py b/ldclient/impl/integrations/files/file_data_source.py index 0fd0593c..81fb38d3 100644 --- a/ldclient/impl/integrations/files/file_data_source.py +++ b/ldclient/impl/integrations/files/file_data_source.py @@ -183,7 +183,7 @@ def __init__(self, resolved_paths, reloader, interval): self._paths = resolved_paths self._reloader = reloader self._file_times = self._check_file_times() - self._timer = RepeatingTask("ldclient.datasource.file.poll", interval, interval, self._poll) + self._timer = RepeatingTask.at_interval("ldclient.datasource.file.poll", interval, interval, self._poll) self._timer.start() def stop(self): diff --git a/ldclient/impl/integrations/files/file_data_sourcev2.py b/ldclient/impl/integrations/files/file_data_sourcev2.py index 5442b81e..032fc3fb 100644 --- a/ldclient/impl/integrations/files/file_data_sourcev2.py +++ b/ldclient/impl/integrations/files/file_data_sourcev2.py @@ -398,7 +398,7 @@ def __init__(self, resolved_paths, on_change_callback, interval): self._paths = resolved_paths self._on_change = on_change_callback self._file_times = self._check_file_times() - self._timer = RepeatingTask( + self._timer = RepeatingTask.at_interval( "ldclient.datasource.filev2.poll", interval, interval, self._poll ) self._timer.start() diff --git a/ldclient/impl/repeating_task.py b/ldclient/impl/repeating_task.py index 2d65de87..1b58c481 100644 --- a/ldclient/impl/repeating_task.py +++ b/ldclient/impl/repeating_task.py @@ -1,39 +1,84 @@ -import time from threading import Event, Thread -from typing import Callable +from typing import Any, Callable, Protocol from ldclient.impl.util import log +class DelaySource(Protocol): + """Supplies the wait before a repeating task's next invocation.""" + + @property + def next_delay(self) -> float: + """The seconds to wait before the next invocation.""" + ... + + +class FixedDelay(DelaySource): + """A :class:`DelaySource` that always gives the same wait.""" + + def __init__(self, seconds: float): + self.__seconds = seconds + + @property + def next_delay(self) -> float: + return self.__seconds + + class RepeatingTask: """ - A generic mechanism for calling a callback repeatedly at fixed intervals on a worker thread. + A generic mechanism for calling a callback repeatedly on a worker thread. + + The wait between invocations comes from a :class:`DelaySource`, which the + task reads after each one. Use :meth:`at_interval` for the common case of + a fixed interval. """ - def __init__(self, label, interval: float, initial_delay: float, callable: Callable): + def __init__(self, label: str, delays: DelaySource, initial_delay: float, callable: Callable[[], Any]): """ Creates the task, but does not start the worker thread yet. - :param interval: maximum time in seconds between invocations of the callback + :param label: names the worker thread, and appears in log messages + :param delays: supplies the wait after each invocation returns :param initial_delay: time in seconds to wait before the first invocation - :param callable: the function to execute repeatedly + :param callable: the function to execute repeatedly. Anything it + returns is ignored. """ - self.__interval = interval + self.__label = label + self.__delays = delays self.__initial_delay = initial_delay self.__action = callable self.__stop = Event() + self.__started = False self.__thread = Thread(target=self._run, name=f"{label}.repeating") self.__thread.daemon = True + @staticmethod + def at_interval(label: str, interval: float, initial_delay: float, callable: Callable[[], Any]) -> 'RepeatingTask': + """ + Creates a task that runs at a fixed interval. + + :param interval: time in seconds to wait after each invocation returns + """ + return RepeatingTask(label, FixedDelay(interval), initial_delay, callable) + def start(self): """ - Starts the worker thread. + Starts the worker thread, if it is not running already. + + Starting a task twice logs and does nothing, rather than raising, so a + caller that is safe to call more than once stays safe. """ + if self.__started: + log.info("Task %s has already been started; ignoring" % self.__label) + return + self.__started = True self.__thread.start() def stop(self): """ - Tells the worker thread to stop. It cannot be restarted after this. + Tells the worker thread to stop. + + The stop is permanent. A later :meth:`start` does not resume the task. """ self.__stop.set() @@ -43,10 +88,11 @@ def _run(self): return stopped = self.__stop.is_set() while not stopped: - next_time = time.time() + self.__interval try: self.__action() except Exception as e: log.exception("Unexpected exception on worker thread: %s" % e) - delay = next_time - time.time() + # 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() diff --git a/ldclient/impl/retry.py b/ldclient/impl/retry.py new file mode 100644 index 00000000..ccba064a --- /dev/null +++ b/ldclient/impl/retry.py @@ -0,0 +1,342 @@ +""" +Computes how long to wait before a failed operation is tried again. + +Each failure falls into one of two classes. A ``NORMAL`` failure is one the +service is expected to recover from soon, so the wait stays short. An +``UNEXPECTED`` failure points to a problem that a person has to fix, such as a +rejected SDK key, so the wait becomes much longer. Only an HTTP status can be +``UNEXPECTED``; every network and TLS failure is ``NORMAL``. Neither class ever +tells the caller to give up. There is always a next attempt. + +The wait doubles after each failure, up to a ceiling. A random jitter is then +subtracted, so that many callers do not all try again at the same moment. The +wait never falls below the caller's operating cadence. + +:class:`RetryState` holds the state for one caller. Use :func:`for_streaming` +or :func:`for_polling` to build one with the right parameters and reset policy. +""" + +# currently excluded from documentation - see docs/README.md + +import random +import time +from enum import Enum +from typing import Optional, Protocol + +from ldclient.impl.util import log + +# The delay bounds of the extended regime, in seconds. A component enters the +# extended regime after an unexpected failure. +EXTENDED_INITIAL_DELAY = 5 * 60 +EXTENDED_MAX_DELAY = 60 * 60 + +# The delay bounds of the normal regime for streaming, in seconds. The initial +# delay is configurable as ``initial_reconnect_delay``. +STREAMING_MAX_DELAY = 30 + +# The documented default for ``initial_reconnect_delay``, in seconds. It stands +# in for a configured value of zero or less, which would reconnect with no wait +# at all. +DEFAULT_INITIAL_RECONNECT_DELAY = 1 + +# How long streaming must operate without a failure before its retry state +# resets, in seconds. +STREAMING_RESET_INTERVAL = 60 + +# How many polls in a row must succeed before polling's retry state resets. +POLLING_RESET_SUCCESSES = 2 + +# HTTP statuses in the 4xx range that are still normal failures. Every other +# 4xx is unexpected. +_NORMAL_4XX_STATUSES = frozenset([400, 408, 429]) + +# An upper bound on the backoff exponent, so a long outage cannot overflow the +# delay computation. Any real ceiling is reached long before this. +_MAX_BACKOFF_EXPONENT = 30 + + +class FailureKind(Enum): + """How a failure is classified, which decides how long the next wait is.""" + + NORMAL = 'normal' + """A failure the service is expected to recover from without help.""" + + UNEXPECTED = 'unexpected' + """A failure that suggests a problem a person has to fix. The component + keeps retrying, but much less often.""" + + +def classify_http_status(status: int) -> FailureKind: + """ + Classifies an HTTP status. + + ``400``, ``408`` and ``429`` are normal, as is any ``5xx``. Every other + ``4xx`` -- including ``401`` and ``403`` -- is unexpected. + """ + if 400 <= status < 500 and status not in _NORMAL_4XX_STATUSES: + return FailureKind.UNEXPECTED + return FailureKind.NORMAL + + +class ResetPolicy(Protocol): + """Decides when a component has operated well enough for long enough that + its retry state should reset. This is the only behavioral difference + between streaming and polling.""" + + def note_healthy(self) -> None: + """Records that the component is operating normally.""" + ... + + def note_failure(self) -> None: + """Records a failure, which ends any healthy stretch in progress.""" + ... + + def is_satisfied(self) -> bool: + """Reports whether the reset condition is met.""" + ... + + +class AfterHealthyFor(ResetPolicy): + """Resets once the component has operated without failing for + ``seconds``. This is the streaming policy.""" + + def __init__(self, seconds: float): + self._healthy_seconds = seconds + self._healthy_since: Optional[float] = None + + def note_healthy(self) -> None: + """Records the monotonic time the component became healthy. Calling + this again while it is still healthy does not move that time.""" + if self._healthy_since is None: + self._healthy_since = time.monotonic() + + def note_failure(self) -> None: + self._healthy_since = None + + def is_satisfied(self) -> bool: + if self._healthy_since is None: + return False + return time.monotonic() - self._healthy_since >= self._healthy_seconds + + @property + def healthy_since(self) -> Optional[float]: + """When the current healthy stretch began, or None if the component is + not currently healthy.""" + return self._healthy_since + + +class AfterConsecutiveSuccesses(ResetPolicy): + """Resets once ``count`` operations in a row have succeeded. This is the + polling policy.""" + + def __init__(self, count: int): + self._count = count + self._successes = 0 + + def note_healthy(self) -> None: + self._successes += 1 + + def note_failure(self) -> None: + self._successes = 0 + + def is_satisfied(self) -> bool: + return self._successes >= self._count + + @property + def successes(self) -> int: + """How many operations have succeeded in a row.""" + return self._successes + + +class RetryState: + """ + Tracks how long a data source should wait before its next attempt. + + A failure moves the state on and returns the wait. The delay for attempt + ``n`` is ``min(min_delay * 2 ** (n - 1), max_delay)``, less a random + jitter of up to half of it, and never less than the operating cadence. + + An unexpected failure moves the state to the extended regime, which raises + both delay bounds. The bounds stay raised until the reset condition is met, + so a normal failure that follows cannot lower them. + """ + + def __init__( + self, + initial_delay: float, + normal_ceiling: float, + extended_initial_delay: float, + extended_ceiling: float, + reset_policy: ResetPolicy, + operating_cadence: float = 0, + ): + """ + :param initial_delay: the delay before the first retry, in seconds + :param normal_ceiling: the longest normal-regime delay, in seconds + :param extended_initial_delay: the delay before the first retry in the + extended regime, in seconds + :param extended_ceiling: the longest extended-regime delay, in seconds + :param reset_policy: decides when the retry state resets + :param operating_cadence: the rate the component normally operates at, + in seconds; no wait is ever shorter than this. Zero disables the + floor, which is what streaming wants. + """ + self._initial_delay = initial_delay + self._normal_ceiling = normal_ceiling + self._extended_initial_delay = extended_initial_delay + self._extended_ceiling = extended_ceiling + self._reset_policy = reset_policy + self._operating_cadence = operating_cadence + + self._n = 0 + self._extended = False + self._min_delay = initial_delay + self._max_delay = max(normal_ceiling, initial_delay) + self._attempts = 0 + # Read before any outcome is recorded, this is the ordinary interval. + self._next_delay = operating_cadence if operating_cadence > 0 else initial_delay + + @property + def next_delay(self) -> float: + """The wait before the next attempt, in seconds, as the last recorded + outcome decided it.""" + return self._next_delay + + @property + def attempts(self) -> int: + """How many failures this state has seen. For logging only.""" + return self._attempts + + @property + def min_delay(self) -> float: + """The delay the current regime starts from, in seconds.""" + return self._min_delay + + @property + def max_delay(self) -> float: + """The longest delay the current regime allows, in seconds.""" + return self._max_delay + + @property + def operating_cadence(self) -> float: + """The rate the component normally operates at, in seconds.""" + return self._operating_cadence + + @property + def in_extended_regime(self) -> bool: + """Whether an unexpected failure has moved this state to the extended + delay bounds.""" + return self._extended + + def record_failure(self, kind: FailureKind, wait_override: Optional[float] = None) -> float: + """ + Records a failed attempt and returns how long to wait before the next + one, in seconds. + + The state moves on before the wait is computed, so the wait always + reflects the failure just recorded. + + :param kind: how the failure was classified + :param wait_override: a wait the server asked for, which replaces the + computed one. LaunchDarkly does not send one on these endpoints, + so this is an unused seam. + """ + # Only a time-based policy needs this: nothing runs while a stream is healthy. + self._reset_if_due() + self._attempts += 1 + self._reset_policy.note_failure() + + if kind is FailureKind.UNEXPECTED and not self._extended: + # Moving to the extended regime raises both bounds and starts the + # attempt count over. Only the move does this: a later unexpected + # failure keeps counting up, so the delay is not pinned to the + # extended initial delay. + self._extended = True + self._min_delay = self._extended_initial_delay + self._max_delay = max(self._extended_ceiling, self._min_delay) + self._n = 1 + else: + self._n += 1 + + self._next_delay = self._compute_wait(wait_override) + return self._next_delay + + def record_success(self) -> None: + """ + Records a successful operation, and resets the retry state if that is + now enough. + + The wait before the next operation becomes the operating cadence, even + when the retry state is still raised, because a backoff wait applies to + a retry and not to every operation. + """ + self._reset_policy.note_healthy() + self._reset_if_due() + self._next_delay = self._operating_cadence + + def _reset_if_due(self) -> None: + """Clears the retry state when the reset policy is satisfied, returning + the delay bounds to the normal regime.""" + if not self._reset_policy.is_satisfied(): + return + self._n = 0 + self._extended = False + self._min_delay = self._initial_delay + self._max_delay = max(self._normal_ceiling, self._initial_delay) + + def _compute_wait(self, wait_override: Optional[float]) -> float: + if wait_override is not None: + return max(wait_override, self._operating_cadence) + exponent = min(max(self._n - 1, 0), _MAX_BACKOFF_EXPONENT) + delay = min(self._min_delay * (2**exponent), self._max_delay) + jitter = random.random() * delay / 2 + return max(delay - jitter, self._operating_cadence) + + +def for_streaming(initial_reconnect_delay: float) -> RetryState: + """ + Builds the retry state for a streaming data source. + + Streaming has no operating cadence, so there is no floor on the wait. It + is healthy from the first message of a fresh stream, and resets after a + minute of that. + + A configured delay of zero or less would reconnect with no wait, so the + documented default stands in for it. ``Config`` does not check this value, + though it does clamp ``poll_interval``. + + The extended regime never starts below the configured delay. + """ + if initial_reconnect_delay <= 0: + log.warning( + "initial_reconnect_delay must be greater than zero; using the default of %ss" + % DEFAULT_INITIAL_RECONNECT_DELAY + ) + initial_reconnect_delay = DEFAULT_INITIAL_RECONNECT_DELAY + return RetryState( + initial_delay=initial_reconnect_delay, + normal_ceiling=STREAMING_MAX_DELAY, + extended_initial_delay=max(EXTENDED_INITIAL_DELAY, initial_reconnect_delay), + extended_ceiling=EXTENDED_MAX_DELAY, + reset_policy=AfterHealthyFor(STREAMING_RESET_INTERVAL), + ) + + +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. + """ + return RetryState( + initial_delay=poll_interval, + normal_ceiling=poll_interval, + extended_initial_delay=max(EXTENDED_INITIAL_DELAY, poll_interval), + extended_ceiling=EXTENDED_MAX_DELAY, + reset_policy=AfterConsecutiveSuccesses(POLLING_RESET_SUCCESSES), + operating_cadence=poll_interval, + ) diff --git a/ldclient/impl/util.py b/ldclient/impl/util.py index 69e7186a..63e90624 100644 --- a/ldclient/impl/util.py +++ b/ldclient/impl/util.py @@ -134,9 +134,15 @@ def throw_if_unsuccessful_response(resp): def is_http_error_recoverable(status): + """ + Reports whether a component that treats some statuses as fatal should + keep going. + + 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 + return status in _RETRYABLE_STATUSES # all other 4xx besides these are treated as fatal + return True def http_error_description(status): @@ -144,6 +150,13 @@ def http_error_description(status): def http_error_message(status, context, retryable_message="will retry"): + """ + Builds the log message for an HTTP failure in a component that stops on + some statuses. + + Deprecated. The FDv1 data sources build their own message instead, 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..04aed18c 100644 --- a/ldclient/interfaces.py +++ b/ldclient/interfaces.py @@ -1005,16 +1005,15 @@ 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. - 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 means the SDK client was explicitly shut down, or that its configuration could not be parsed. """ diff --git a/ldclient/testing/impl/datasource/test_async_polling.py b/ldclient/testing/impl/datasource/test_async_polling.py index c639efb2..8a87a6dc 100644 --- a/ldclient/testing/impl/datasource/test_async_polling.py +++ b/ldclient/testing/impl/datasource/test_async_polling.py @@ -3,9 +3,13 @@ """ import asyncio +import logging +import ssl 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 +17,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 +30,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 +44,34 @@ 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, +) + + +def fast_retry_state(delay=0.001): + """A retry state with tiny delays, so a test does not have to wait out the + real extended-regime delay of five minutes.""" + return RetryState( + initial_delay=delay, + normal_ceiling=delay, + extended_initial_delay=delay, + extended_ceiling=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 +86,7 @@ def make_processor(config=None, store=None, ready=None, requester=None): requester=requester, store=store, ready=ready, + retry_state=retry_state, ) @@ -174,31 +213,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.in_extended_regime await processor.stop() + @pytest.mark.asyncio + async def test_the_first_success_after_an_outage_polls_at_the_cadence(self): + # RETRY 1.4.8: a backoff wait applies to a retry, not to every + # operation. _poll returns the wait, so this reads it directly 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.in_extended_regime, "one success restores the cadence but does not reset" + + await processor._fetch_and_store() + assert retry.next_delay == 30 + assert not retry.in_extended_regime, "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): @@ -291,6 +368,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.in_extended_regime + + @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() @@ -357,7 +497,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 +507,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 +515,65 @@ 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 + @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 @@ -426,16 +616,19 @@ async def test_initialized_returns_false_before_first_poll(self): @pytest.mark.asyncio @patch('ldclient.config.Config.poll_interval', new_callable=MagicMock) - async def test_second_start_call_raises(self, mock_interval): + async def test_second_start_call_is_a_no_op(self, mock_interval): + # AsyncLDClient.start() is documented as an idempotent no-op, so + # nothing underneath it may raise on a repeat call. mock_interval.__get__ = MagicMock(return_value=0) processor = make_processor() processor._requester.get_all_data = AsyncMock(return_value=SAMPLE_DATA) processor.start() - # Like a thread, the polling task can only be started once - with pytest.raises(RuntimeError): - processor.start() + first_task = processor._task + processor.start() + + assert processor._task is first_task await processor.stop() @@ -448,15 +641,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_streaming.py b/ldclient/testing/impl/datasource/test_async_streaming.py index ffd5a9ea..a1ef0122 100644 --- a/ldclient/testing/impl/datasource/test_async_streaming.py +++ b/ldclient/testing/impl/datasource/test_async_streaming.py @@ -7,9 +7,13 @@ import asyncio import json +import logging +import ssl 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 +21,23 @@ AsyncStreamingUpdateProcessor ) from ldclient.impl.model import ModelEntity +from ldclient.impl.retry import ( + EXTENDED_INITIAL_DELAY, + EXTENDED_MAX_DELAY, + STREAMING_MAX_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 +93,57 @@ 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( + initial_delay=0.001, + normal_ceiling=0.001, + extended_initial_delay=0.001, + extended_ceiling=0.001, + reset_policy=policy, + ) + + +def _fast_retry_state(delay: float = 0.001) -> RetryState: + """A retry state with tiny delays, so a test does not have to wait out the + real extended-regime delay of five minutes.""" + return RetryState( + initial_delay=delay, + normal_ceiling=delay, + extended_initial_delay=delay, + extended_ceiling=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( + initial_delay=0, + normal_ceiling=STREAMING_MAX_DELAY, + extended_initial_delay=EXTENDED_INITIAL_DELAY, + extended_ceiling=EXTENDED_MAX_DELAY, + reset_policy=AfterHealthyFor(STREAMING_RESET_INTERVAL), + ) + + class _MockSSE: """Stand-in for AsyncSSEClient exposing the surface the processor uses.""" @@ -101,31 +170,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 +294,111 @@ 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.in_extended_regime 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_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.in_extended_regime + 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_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 +416,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 +435,166 @@ 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.in_extended_regime) + + 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.in_extended_regime + + 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.in_extended_regime + assert retry.max_delay == STREAMING_MAX_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_stop_closes_sse_and_finishes_task(): flag = FlagBuilder('f1').version(1).build() @@ -295,13 +612,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() diff --git a/ldclient/testing/impl/datasource/test_polling_processor.py b/ldclient/testing/impl/datasource/test_polling_processor.py index 06e92d89..9d073167 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,7 @@ ) from ldclient.testing.builders import * from ldclient.testing.stub_util import MockFeatureRequester, MockResponse -from ldclient.testing.test_util import SpyListener +from ldclient.testing.test_util import SpyListener, no_retry_jitter from ldclient.versioned_data_kind import FEATURES, SEGMENTS pp = None @@ -37,9 +46,22 @@ def teardown_function(): pp.stop() -def setup_processor(config): +def fast_retry_state(delay=0.05): + """A retry state with tiny delays, so a test does not have to wait out the + real extended-regime delay of five minutes.""" + return RetryState( + initial_delay=delay, + normal_ceiling=delay, + extended_initial_delay=delay, + extended_ceiling=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 +99,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_causes_immediate_failure(): - verify_unrecoverable_http_error(401) +def test_http_403_error_does_not_stop_polling(): + verify_unexpected_http_error(403) + + +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 +128,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 +140,228 @@ 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. + assert not ready.wait(0.4) + assert retry.in_extended_regime 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(): + # RETRY 1.4.8: a backoff wait applies to a retry, not to every operation. + # _poll returns the wait, so this reads it directly 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.in_extended_regime, "one success restores the cadence but does not reset" + + processor._poll() + assert retry.next_delay == 30 + assert not retry.in_extended_regime, "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.in_extended_regime + + +@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.in_extended_regime, "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_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_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..39f69875 100644 --- a/ldclient/testing/impl/datasource/test_streaming.py +++ b/ldclient/testing/impl/datasource/test_streaming.py @@ -1,15 +1,32 @@ +import logging +import ssl import time from threading import Event 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 ( + STREAMING_RESET_INTERVAL, + AfterHealthyFor, + RetryState, + for_streaming +) from ldclient.interfaces import ( DataSourceErrorKind, DataSourceState, @@ -30,12 +47,30 @@ make_put_event, stream_content ) -from ldclient.testing.test_util import SpyListener +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 + +def fast_retry_state(delay=brief_delay): + """A retry state with tiny delays, so a test does not have to wait out the + real extended-regime delay of five minutes.""" + return RetryState( + initial_delay=delay, + normal_ceiling=delay, + extended_initial_delay=delay, + extended_ceiling=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 # described at https://stackoverflow.com/questions/2617615/slow-python-http-server-on-localhost but we had no @@ -257,7 +292,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 +306,344 @@ def test_unrecoverable_http_error(status): with StreamingUpdateProcessor(config, store, ready, None) as sp: sp.start() - ready.wait(5) + # Initialization is not falsely unblocked: the caller waits out + # its own start_wait and then finds the client uninitialized. + assert not ready.wait(1) assert not sp.initialized() + assert sp.is_alive() + assert sp._retry.in_extended_regime 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.in_extended_regime + + 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.in_extended_regime + 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 _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_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( + initial_delay=brief_delay, + normal_ceiling=brief_delay, + extended_initial_delay=brief_delay, + extended_ceiling=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( + initial_delay=brief_delay, + normal_ceiling=brief_delay, + extended_initial_delay=brief_delay, + extended_ceiling=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.in_extended_regime + assert retry.max_delay == 30 + + def test_http_proxy(monkeypatch): def _stream_processor_proxy_test(server, config, secure): store = InMemoryFeatureStore() @@ -407,6 +777,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 +798,15 @@ def test_failure_transitions_from_valid(): with StreamingUpdateProcessor(config, store, ready, None) as sp: sp.start() - ready.wait(start_wait) + # The 401 is retried five minutes out, so readiness never fires. + assert not ready.wait(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_repeating_task.py b/ldclient/testing/impl/test_repeating_task.py index 7d29cbf3..0fdf89ff 100644 --- a/ldclient/testing/impl/test_repeating_task.py +++ b/ldclient/testing/impl/test_repeating_task.py @@ -1,13 +1,14 @@ +import logging import time from queue import Empty, Queue from threading import Event -from ldclient.impl.repeating_task import RepeatingTask +from ldclient.impl.repeating_task import DelaySource, FixedDelay, RepeatingTask def test_task_does_not_start_when_created(): signal = Event() - task = RepeatingTask("ldclient.testing.set-signal", 0.01, 0, lambda: signal.set()) + task = RepeatingTask.at_interval("ldclient.testing.set-signal", 0.01, 0, lambda: signal.set()) try: signal_was_set = signal.wait(0.1) assert signal_was_set is False @@ -15,9 +16,47 @@ def test_task_does_not_start_when_created(): task.stop() +def test_a_second_start_logs_and_does_not_raise(caplog): + """A raise here can surface out of a caller that is documented as safe to + call more than once, such as AsyncLDClient.start().""" + caplog.set_level(logging.INFO) + queue = Queue() + task = RepeatingTask.at_interval("ldclient.testing.enqueue-time", 0.01, 0, lambda: queue.put(time.time())) + try: + task.start() + thread = task._RepeatingTask__thread + + task.start() + + assert task._RepeatingTask__thread is thread + assert queue.get(True, 1) is not None # still running + finally: + task.stop() + + assert any( + r.getMessage() == "Task ldclient.testing.enqueue-time has already been started; ignoring" + for r in caplog.records + ) + + +def test_a_start_after_stop_does_not_resume_the_task(): + counter = 0 + + def do_task(): + nonlocal counter + counter += 1 + + task = RepeatingTask.at_interval("ldclient.testing.task-runner", 0.01, 0, do_task) + task.stop() + task.start() + time.sleep(0.1) + + assert counter == 0 + + def test_task_executes_until_stopped(): queue = Queue() - task = RepeatingTask("ldclient.testing.enqueue-time", 0.1, 0, lambda: queue.put(time.time())) + task = RepeatingTask.at_interval("ldclient.testing.enqueue-time", 0.1, 0, lambda: queue.put(time.time())) try: last = None task.start() @@ -39,6 +78,59 @@ def test_task_executes_until_stopped(): assert no_more_items is True +class _MutableDelay(DelaySource): + """A delay source a test can move between invocations.""" + + def __init__(self, seconds: float): + self.seconds = seconds + + @property + def next_delay(self) -> float: + return self.seconds + + +def test_fixed_delay_always_gives_the_same_wait(): + delays = FixedDelay(2.5) + assert delays.next_delay == 2.5 + assert delays.next_delay == 2.5 + + +def test_the_task_reads_the_delay_source_after_every_invocation(): + """A value the action decides takes effect on the next wait.""" + reads = Queue() + delays = _MutableDelay(0.01) + + def do_task(): + reads.put(delays.seconds) + delays.seconds = 0.02 # what the next wait must use + + task = RepeatingTask("ldclient.testing.mutable-delay", delays, 0, do_task) + try: + task.start() + assert reads.get(True, 1) == 0.01 + assert reads.get(True, 1) == 0.02 + assert reads.get(True, 1) == 0.02 + finally: + task.stop() + + +def test_whatever_the_action_returns_is_ignored(): + """Guards big-segment polling, whose action returns a status object.""" + calls = Queue() + + def do_task(): + calls.put(time.time()) + return object() # not a number, and not for the task to interpret + + task = RepeatingTask.at_interval("ldclient.testing.returns-a-value", 0.01, 0, do_task) + try: + task.start() + for _ in range(3): + assert calls.get(True, 1) is not None + finally: + task.stop() + + def test_task_can_be_stopped_from_within_the_task(): counter = 0 stopped = Event() @@ -51,7 +143,7 @@ def do_task(): task.stop() stopped.set() - task = RepeatingTask("ldclient.testing.task-runner", 0.01, 0, do_task) + task = RepeatingTask.at_interval("ldclient.testing.task-runner", 0.01, 0, do_task) try: task.start() assert stopped.wait(0.1) is True diff --git a/ldclient/testing/impl/test_retry.py b/ldclient/testing/impl/test_retry.py new file mode 100644 index 00000000..d82f2569 --- /dev/null +++ b/ldclient/testing/impl/test_retry.py @@ -0,0 +1,471 @@ +""" +Tests for ldclient.impl.retry. + +Nothing here sleeps. A test that needs to move time on uses ``frozen_clock``, +which replaces the ``time`` module the retry module reads. Jitter is removed +for every test by an autouse fixture, so a delay assertion reads the +undisturbed value; the tests that are about jitter override it. +""" + +import logging +import random +from contextlib import contextmanager +from unittest import mock + +import pytest + +from ldclient.impl import retry +from ldclient.impl.retry import ( + DEFAULT_INITIAL_RECONNECT_DELAY, + EXTENDED_INITIAL_DELAY, + EXTENDED_MAX_DELAY, + POLLING_RESET_SUCCESSES, + STREAMING_MAX_DELAY, + STREAMING_RESET_INTERVAL, + AfterConsecutiveSuccesses, + AfterHealthyFor, + FailureKind, + RetryState, + classify_http_status, + for_polling, + for_streaming +) +from ldclient.testing.test_util import fixed_retry_jitter, no_retry_jitter + +NORMAL = FailureKind.NORMAL +UNEXPECTED = FailureKind.UNEXPECTED + + +class _FrozenClock: + """Stands in for the ``time`` module. Time only moves when a test says so.""" + + def __init__(self, now: float): + self.now = now + + def monotonic(self) -> float: + return self.now + + def advance(self, seconds: float) -> None: + self.now += seconds + + +@contextmanager +def frozen_clock(now: float = 1000.0): + """Freezes the clock the retry module reads. + + Patching the module's own ``time`` reference keeps the change local to + ``ldclient.impl.retry``; every other module keeps the real clock. + """ + clock = _FrozenClock(now) + with mock.patch.object(retry, 'time', clock): + yield clock + + +# The random draw just below 1, which subtracts as much jitter as the spec +# allows: half the delay. +FULL_JITTER = 0.9999999 + + +@pytest.fixture(autouse=True) +def without_jitter(): + """Removes jitter for every test in this module, so a delay assertion can + read the undisturbed value.""" + with no_retry_jitter(): + yield + + +@contextmanager +def real_jitter(): + """Restores the real random source, for a test that asserts the bounds hold + for any draw rather than for one fixed value.""" + with mock.patch.object(retry, 'random', random): + yield + + +def streaming_state(initial_delay=1): + return for_streaming(initial_delay) + + +def polling_state(poll_interval=30): + return for_polling(poll_interval) + + +class TestClassifyHttpStatus: + @pytest.mark.parametrize("status", [400, 408, 429]) + def test_retryable_4xx_statuses_are_normal(self, status): + assert classify_http_status(status) is NORMAL + + @pytest.mark.parametrize("status", [401, 403, 404, 405, 418, 499]) + def test_other_4xx_statuses_are_unexpected(self, status): + assert classify_http_status(status) is UNEXPECTED + + @pytest.mark.parametrize("status", [500, 502, 503, 504, 599]) + def test_5xx_statuses_are_normal(self, status): + assert classify_http_status(status) is NORMAL + + @pytest.mark.parametrize("status", [200, 204, 301, 399]) + def test_non_error_statuses_are_normal(self, status): + assert classify_http_status(status) is NORMAL + + +class TestStreamingInitialDelayGuard: + """``Config`` does not check ``initial_reconnect_delay``, and a value of + zero would reconnect with no wait at all.""" + + @pytest.mark.parametrize("configured", [0, -1, -0.5]) + def test_a_non_positive_delay_falls_back_to_the_default(self, configured, caplog): + caplog.set_level(logging.WARNING) + + state = for_streaming(configured) + + assert state.min_delay == DEFAULT_INITIAL_RECONNECT_DELAY + assert state.record_failure(NORMAL) == DEFAULT_INITIAL_RECONNECT_DELAY + assert caplog.records[0].getMessage() == ( + "initial_reconnect_delay must be greater than zero; using the default of 1s" + ) + + @pytest.mark.parametrize("configured", [0.001, 0.5, 1, 5, 45]) + def test_a_positive_delay_is_left_alone(self, configured, caplog): + caplog.set_level(logging.WARNING) + + state = for_streaming(configured) + + assert state.min_delay == configured + assert state.record_failure(NORMAL) == configured + assert caplog.records == [] + + +class TestStreamingExtendedDelayFloor: + """RETRY 1.5.4.1: a delay that applies after an unexpected failure must not + be below the component's initial delay.""" + + @pytest.mark.parametrize( + "configured,expected", + [(1, 300), (30, 300), (300, 300), (600, 600), (3600, 3600), (0, 300), (-5, 300)], + ) + def test_the_extended_delay_never_starts_below_the_configured_delay(self, configured, expected): + assert for_streaming(configured).record_failure(UNEXPECTED) == expected + + @pytest.mark.parametrize("configured", [1, 30, 300, 600, 3600]) + def test_an_unexpected_failure_never_waits_less_than_a_normal_one(self, configured): + normal = for_streaming(configured).record_failure(NORMAL) + unexpected = for_streaming(configured).record_failure(UNEXPECTED) + assert unexpected >= normal + + @pytest.mark.parametrize( + "configured,ladder", + [ + (1, [300, 600, 1200, 2400, 3600]), + (600, [600, 1200, 2400, 3600, 3600]), + ], + ids=["default", "clamped"], + ) + def test_the_extended_ladder_still_doubles_to_the_ceiling(self, configured, ladder): + state = for_streaming(configured) + delays = [state.record_failure(UNEXPECTED)] + delays += [state.record_failure(NORMAL) for _ in range(4)] + assert delays == ladder + + +class TestStreamingDelayTable: + def test_normal_regime_doubles_up_to_the_ceiling(self): + state = streaming_state(initial_delay=1) + delays = [state.record_failure(NORMAL) for _ in range(8)] + assert delays == [1, 2, 4, 8, 16, 30, 30, 30] + + def test_extended_regime_doubles_up_to_the_ceiling(self): + state = streaming_state(initial_delay=1) + delays = [state.record_failure(UNEXPECTED)] + delays += [state.record_failure(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): + # RETRY 1.5.4 as amended: maxDelay must not fall below initialDelay. + state = streaming_state(initial_delay=45) + assert state.max_delay == 45 + assert state.record_failure(NORMAL) == 45 + + def test_the_ceiling_is_sticky_once_the_extended_regime_starts(self): + # RETRY 1.5.5: a normal failure after an unexpected one must not lower + # the bounds back to the normal regime. + state = streaming_state(initial_delay=1) + state.record_failure(UNEXPECTED) + assert state.in_extended_regime + assert state.max_delay == EXTENDED_MAX_DELAY + + state.record_failure(NORMAL) + assert state.in_extended_regime + assert state.max_delay == EXTENDED_MAX_DELAY + assert state.min_delay == EXTENDED_INITIAL_DELAY + + def test_a_second_unexpected_failure_keeps_counting_up(self): + # Restarting the count on every unexpected failure would pin the delay + # at the extended initial delay for ever. + state = streaming_state(initial_delay=1) + assert state.record_failure(UNEXPECTED) == 5 * 60 + assert state.record_failure(UNEXPECTED) == 10 * 60 + assert state.record_failure(UNEXPECTED) == 20 * 60 + + def test_the_streaming_defaults_match_the_spec(self): + state = streaming_state(initial_delay=1) + assert state.max_delay == STREAMING_MAX_DELAY + assert state.operating_cadence == 0 + assert STREAMING_RESET_INTERVAL == 60 + + +class TestJitter: + def test_jitter_never_removes_more_than_half_the_delay(self): + with fixed_retry_jitter(FULL_JITTER): + state = streaming_state(initial_delay=8) + delay = state.record_failure(NORMAL) + assert 4 <= delay < 8 + + def test_no_jitter_leaves_the_delay_alone(self): + state = streaming_state(initial_delay=8) + assert state.record_failure(NORMAL) == 8 + + def test_every_delay_stays_within_the_jitter_bounds(self): + # The real random source, so the bound has to hold for any draw rather + # than for one seeded sequence. + with real_jitter(): + state = streaming_state(initial_delay=1) + for base in [1, 2, 4, 8, 16, 30, 30, 30]: + delay = state.record_failure(NORMAL) + assert base / 2 <= delay <= base + + +class TestStreamingReset: + def test_a_minute_of_healthy_operation_resets_the_state(self): + # RETRY 1.8.2. The whole minute passes instantly. + with frozen_clock() as clock: + state = streaming_state(initial_delay=1) + state.record_failure(UNEXPECTED) + state.record_failure(NORMAL) + + state.record_success() + assert state.in_extended_regime, "the window has not elapsed yet" + + clock.advance(STREAMING_RESET_INTERVAL) + state.record_success() + assert not state.in_extended_regime + assert state.max_delay == STREAMING_MAX_DELAY + assert state.record_failure(NORMAL) == 1 + + def test_a_reset_also_happens_on_the_failure_that_ends_a_healthy_stretch(self): + with frozen_clock() as clock: + state = streaming_state(initial_delay=1) + state.record_failure(NORMAL) + state.record_failure(NORMAL) + + state.record_success() + clock.advance(STREAMING_RESET_INTERVAL) + + # The state resets before this failure is counted, so the delay is + # the first-retry delay again rather than the fourth. + assert state.record_failure(NORMAL) == 1 + + def test_a_short_healthy_stretch_does_not_reset(self): + with frozen_clock() as clock: + state = streaming_state(initial_delay=1) + state.record_failure(NORMAL) + + state.record_success() + clock.advance(STREAMING_RESET_INTERVAL - 1) + assert state.record_failure(NORMAL) == 2 + + def test_a_fast_flapping_connection_does_not_ratchet_into_the_extended_regime(self): + # Every transport failure is normal, so no amount of flapping reaches + # the extended regime. Each cycle is a healthy stretch shorter than the + # reset window, so the delay climbs, but only to the normal ceiling. + with frozen_clock() as clock: + state = streaming_state(initial_delay=1) + delays = [] + for _ in range(20): + state.record_success() + clock.advance(5) + delays.append(state.record_failure(NORMAL)) + clock.advance(1) + + assert not state.in_extended_regime + assert max(delays) == STREAMING_MAX_DELAY + assert state.max_delay == STREAMING_MAX_DELAY + + +class TestPollingCadence: + def test_a_normal_failure_polls_again_on_schedule(self): + state = polling_state(poll_interval=30) + assert [state.record_failure(NORMAL) for _ in range(4)] == [30, 30, 30, 30] + + def test_the_extended_regime_doubles_up_to_an_hour(self): + state = polling_state(poll_interval=30) + delays = [state.record_failure(UNEXPECTED)] + delays += [state.record_failure(NORMAL) for _ in range(5)] + assert delays == [5 * 60, 10 * 60, 20 * 60, 40 * 60, 60 * 60, 60 * 60] + + def test_the_wait_never_falls_below_the_poll_interval(self): + # RETRY 1.4.9. Full jitter would otherwise halve the delay. + with fixed_retry_jitter(FULL_JITTER): + state = polling_state(poll_interval=30) + assert state.record_failure(NORMAL) == 30 + assert state.record_failure(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 = polling_state(poll_interval=2 * 60 * 60) + assert state.record_failure(UNEXPECTED) == 2 * 60 * 60 + assert state.max_delay == 2 * 60 * 60 + assert state.min_delay == 2 * 60 * 60 + + def test_one_success_restores_the_cadence_while_the_state_is_still_raised(self): + # RETRY 1.4.8. Conflating this with the reset is the bug another SDK + # shipped: its first successful poll after an outage still waited + # twenty minutes or more. + state = polling_state(poll_interval=30) + state.record_failure(UNEXPECTED) + state.record_failure(NORMAL) + assert state.record_failure(NORMAL) == 20 * 60 + + state.record_success() + assert state.next_delay == 30 + assert state.in_extended_regime, "one success does not reset the state" + + def test_two_successes_in_a_row_reset_the_state(self): + # RETRY 1.8.2 with the polling reset policy. + state = polling_state(poll_interval=30) + state.record_failure(UNEXPECTED) + + state.record_success() + assert state.in_extended_regime + + state.record_success() + assert not state.in_extended_regime + assert state.next_delay == 30 + assert state.record_failure(NORMAL) == 30 + + def test_a_failure_between_two_successes_clears_the_first(self): + state = polling_state(poll_interval=30) + state.record_failure(UNEXPECTED) + state.record_success() + state.record_failure(NORMAL) + state.record_success() + assert state.in_extended_regime + + state.record_success() + assert not state.in_extended_regime + + def test_the_polling_defaults_match_the_spec(self): + state = polling_state(poll_interval=30) + assert state.operating_cadence == 30 + assert state.min_delay == 30 + assert state.max_delay == 30 + assert POLLING_RESET_SUCCESSES == 2 + + +class TestWaitOverride: + def test_an_override_replaces_the_computed_wait(self): + state = streaming_state(initial_delay=1) + state.record_failure(NORMAL) + assert state.record_failure(NORMAL, wait_override=7) == 7 + + def test_an_override_still_respects_the_cadence(self): + state = polling_state(poll_interval=30) + assert state.record_failure(NORMAL, wait_override=1) == 30 + + +class TestAttemptCount: + def test_attempts_counts_every_failure(self): + state = streaming_state(initial_delay=1) + for _ in range(5): + state.record_failure(NORMAL) + assert state.attempts == 5 + + def test_a_reset_does_not_clear_the_attempt_count(self): + # The count is for logging, so it should keep counting across a reset. + with frozen_clock() as clock: + state = streaming_state(initial_delay=1) + state.record_failure(NORMAL) + state.record_success() + clock.advance(STREAMING_RESET_INTERVAL) + state.record_success() + + # The reset shows in the delay dropping back to the first-retry + # value, while the count carries on. + assert state.record_failure(NORMAL) == 1 + assert state.attempts == 2 + + +class TestResetPolicies: + def test_healthy_for_tracks_the_start_of_the_stretch(self): + with frozen_clock() as clock: + policy = AfterHealthyFor(60) + assert not policy.is_satisfied() + + policy.note_healthy() + started = policy.healthy_since + + # A later signal must not push the start of the stretch out. + clock.advance(40) + policy.note_healthy() + assert policy.healthy_since == started + + clock.advance(20) + assert policy.is_satisfied() + + def test_many_healthy_signals_do_not_move_the_window(self): + """Streaming signals on every message, so an unconditional assignment + here would push its reset out for ever -- Go's SDK-2845.""" + with frozen_clock() as clock: + policy = AfterHealthyFor(60) + policy.note_healthy() + first = policy.healthy_since + + for _ in range(59): + clock.advance(1) + policy.note_healthy() + + assert policy.healthy_since == first + assert not policy.is_satisfied() + + # The threshold lands 60s after the first signal, not the last. + clock.advance(1) + policy.note_healthy() + assert policy.is_satisfied() + + def test_healthy_for_is_cleared_by_a_failure(self): + with frozen_clock() as clock: + policy = AfterHealthyFor(60) + policy.note_healthy() + policy.note_failure() + assert policy.healthy_since is None + + clock.advance(900) + assert not policy.is_satisfied() + + def test_consecutive_successes_counts_up(self): + policy = AfterConsecutiveSuccesses(2) + policy.note_healthy() + assert not policy.is_satisfied() + policy.note_healthy() + assert policy.is_satisfied() + + def test_consecutive_successes_is_cleared_by_a_failure(self): + policy = AfterConsecutiveSuccesses(2) + policy.note_healthy() + policy.note_failure() + assert policy.successes == 0 + assert not policy.is_satisfied() + + +class TestLongOutage: + def test_a_long_outage_cannot_overflow_the_delay(self): + state = RetryState( + initial_delay=1, + normal_ceiling=30, + extended_initial_delay=EXTENDED_INITIAL_DELAY, + extended_ceiling=EXTENDED_MAX_DELAY, + reset_policy=AfterHealthyFor(60), + ) + for _ in range(5000): + delay = state.record_failure(NORMAL) + assert delay == 30 diff --git a/ldclient/testing/test_aio.py b/ldclient/testing/test_aio.py index 85174285..94e6d80c 100644 --- a/ldclient/testing/test_aio.py +++ b/ldclient/testing/test_aio.py @@ -6,6 +6,7 @@ """ import asyncio +import logging import subprocess import sys import threading @@ -127,7 +128,7 @@ async def test_async_fires_repeatedly_then_stops(self): async def action(): counts['n'] += 1 - task = aio.AsyncRepeatingTask("test.repeating", 0.01, 0, action) + task = aio.AsyncRepeatingTask.at_interval("test.repeating", 0.01, 0, action) task.start() await _async_wait_until(lambda: counts['n'] >= 3) task.stop() @@ -143,7 +144,7 @@ async def test_async_initial_delay_respected(self): async def action(): counts['n'] += 1 - task = aio.AsyncRepeatingTask("test.repeating", 0.01, 0.1, action) + task = aio.AsyncRepeatingTask.at_interval("test.repeating", 0.01, 0.1, action) task.start() await asyncio.sleep(0.03) assert counts['n'] == 0 @@ -157,7 +158,7 @@ async def action(): counts['n'] += 1 raise RuntimeError("boom") - task = aio.AsyncRepeatingTask("test.repeating", 0.01, 0, action) + task = aio.AsyncRepeatingTask.at_interval("test.repeating", 0.01, 0, action) task.start() await _async_wait_until(lambda: counts['n'] >= 2) task.stop() @@ -171,22 +172,50 @@ async def action(): counts['n'] += 1 holder['task'].stop() - holder['task'] = aio.AsyncRepeatingTask("test.repeating", 0.01, 0, action) + holder['task'] = aio.AsyncRepeatingTask.at_interval("test.repeating", 0.01, 0, action) holder['task'].start() await asyncio.sleep(0.1) assert counts['n'] == 1 @pytest.mark.asyncio - async def test_async_second_start_raises(self): + async def test_async_second_start_logs_and_does_not_raise(self, caplog): + """Mirrors the sync primitive. A raise here can surface out of a caller + that is documented as safe to call more than once.""" + caplog.set_level(logging.INFO) + counts = {'n': 0} + async def action(): - pass + counts['n'] += 1 + + task = aio.AsyncRepeatingTask.at_interval("test.repeating", 0.01, 0, action) + task.start() + handle = task._AsyncRepeatingTask__task - task = aio.AsyncRepeatingTask("test.repeating", 0.01, 0, action) task.start() - with pytest.raises(RuntimeError): - task.start() + + assert task._AsyncRepeatingTask__task is handle + await _async_wait_until(lambda: counts['n'] >= 1) task.stop() + assert any( + r.getMessage() == "Task test.repeating has already been started; ignoring" + for r in caplog.records + ) + + @pytest.mark.asyncio + async def test_async_start_after_stop_does_not_resume_the_task(self): + counts = {'n': 0} + + async def action(): + counts['n'] += 1 + + task = aio.AsyncRepeatingTask.at_interval("test.repeating", 0.01, 0, action) + task.stop() + task.start() + await asyncio.sleep(0.05) + + assert counts['n'] == 0 + class TestBoundedTaskSet: @pytest.mark.asyncio diff --git a/ldclient/testing/test_ldclient_end_to_end.py b/ldclient/testing/test_ldclient_end_to_end.py index 8e608d14..ce523a53 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,18 @@ 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: + started = time.time() + with LDClient(config=config, start_wait=0.5) as client: + elapsed = time.time() - started + assert elapsed >= 0.5 assert not client.is_initialized() assert client.variation(always_true_flag['key'], user, False) is False @@ -91,12 +98,16 @@ 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: + started = time.time() + with LDClient(config=config, start_wait=0.5) as client: + elapsed = time.time() - started + assert elapsed >= 0.5 assert not client.is_initialized() assert client.variation(always_true_flag['key'], user, False) is False diff --git a/ldclient/testing/test_util.py b/ldclient/testing/test_util.py index fbfac15a..2036d097 100644 --- a/ldclient/testing/test_util.py +++ b/ldclient/testing/test_util.py @@ -1,7 +1,10 @@ import os +from contextlib import contextmanager +from unittest import mock import pytest +from ldclient.impl import retry from ldclient.impl.util import redact_password skip_database_tests = os.environ.get('LD_SKIP_DATABASE_TESTS') == '1' @@ -25,6 +28,69 @@ def test_can_redact_password(password_redaction_tests): assert redact_password(input) == expected +class _FixedRandom: + """Stands in for the ``random`` module, always drawing the same value.""" + + def __init__(self, value: float): + self.value = value + + def random(self) -> float: + return self.value + + +class _TickingClock: + """Stands in for the ``time`` module, moving on with every read.""" + + def __init__(self, start: float, step: float): + self.now = start + self.step = step + + def monotonic(self) -> float: + self.now += self.step + return self.now + + +def record_healthy_windows(policy) -> list: + """Records the window each ``note_healthy`` call leaves in place, so a test + can tell one window from the next.""" + windows: list = [] + note = policy.note_healthy + + def wrapper(): + note() + windows.append(policy.healthy_since) + + policy.note_healthy = wrapper # type: ignore[method-assign] + return windows + + +@contextmanager +def ticking_clock(start: float = 1000.0, step: float = 1.0): + """Gives :mod:`ldclient.impl.retry` a clock that moves on every read, so a + timestamp it stored can be told apart from one it stored later.""" + with mock.patch.object(retry, 'time', _TickingClock(start, step)): + yield + + +@contextmanager +def fixed_retry_jitter(fraction: float): + """Fixes the jitter that :mod:`ldclient.impl.retry` subtracts from a delay. + + ``0`` subtracts none, so a test can assert an exact delay. A value just + below ``1`` subtracts as much as the spec allows, which is half. + + Patching the retry module's own ``random`` reference keeps the change + local to that module; every other module keeps the real source. + """ + with mock.patch.object(retry, 'random', _FixedRandom(fraction)): + yield + + +def no_retry_jitter(): + """Removes the retry jitter, so a test can assert an exact delay.""" + return fixed_retry_jitter(0.0) + + class SpyListener: def __init__(self): self._statuses = []