feat: Retry indefinitely after a data source failure instead of stopping permanently in FDv1

fix: Warn and use the documented default for an invalid poll interval or initial reconnect delay

Author: jsonbailey

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(

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'}

ldclient/async_client.py

@@ -389,8 +389,8 @@ async def is_initialized(self) -> bool:
 
         If this returns false, it means that the client has not yet successfully connected to LaunchDarkly.
         It might still be in the process of starting up, or it might be attempting to reconnect after an
-        unsuccessful attempt, or it might have received an unrecoverable error (such as an invalid SDK key)
-        and given up.
+        unsuccessful attempt, or it might have received an error that needs to be fixed (such
+        as an invalid SDK key).
 
         This is a coroutine because determining readiness may query a persistent store.
         """

ldclient/async_config.py

@@ -15,6 +15,8 @@
 from ldclient.config import (
     DEFAULT_BASE_URI,
     DEFAULT_EVENTS_URI,
+    DEFAULT_INITIAL_RECONNECT_DELAY,
+    DEFAULT_POLL_INTERVAL,
     DEFAULT_STREAM_URI,
     GET_LATEST_FEATURES_PATH,
     STREAM_FLAGS_PATH,
@@ -149,11 +151,11 @@ def __init__(
         flush_interval: float = 5,
         stream_uri: str = DEFAULT_STREAM_URI,
         stream: bool = True,
-        initial_reconnect_delay: float = 1,
+        initial_reconnect_delay: float = DEFAULT_INITIAL_RECONNECT_DELAY,
         defaults: dict = {},
         send_events: Optional[bool] = None,
         update_processor_class: Optional[Callable[['AsyncConfig', AsyncFeatureStore, AsyncEvent], AsyncUpdateProcessor]] = None,
-        poll_interval: float = 30,
+        poll_interval: float = DEFAULT_POLL_INTERVAL,
         use_ldd: bool = False,
         feature_store: Optional[AsyncFeatureStore] = None,
         feature_requester_class=None,
@@ -256,7 +258,7 @@ def __init__(
         self.__update_processor_class = update_processor_class
         self.__stream = stream
         self.__initial_reconnect_delay = initial_reconnect_delay
-        self.__poll_interval = max(poll_interval, 30.0)
+        self.__poll_interval = max(poll_interval, DEFAULT_POLL_INTERVAL)
         self.__use_ldd = use_ldd
         self.__feature_store = AsyncInMemoryFeatureStore() if not feature_store else feature_store
         self.__event_processor_class = event_processor_class

ldclient/client.py

@@ -369,10 +369,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
         """

ldclient/config.py

@@ -36,6 +36,11 @@
 DEFAULT_EVENTS_URI = 'https://events.launchdarkly.com'
 DEFAULT_STREAM_URI = 'https://stream.launchdarkly.com'
 
+# Defaults, in seconds, for the two configurable data source intervals. The
+# poll interval is also its own minimum.
+DEFAULT_INITIAL_RECONNECT_DELAY = 1
+DEFAULT_POLL_INTERVAL = 30
+
 
 class BigSegmentsConfig:
     """Configuration options related to Big Segments.
@@ -295,11 +300,11 @@ def __init__(
         flush_interval: float = 5,
         stream_uri: str = DEFAULT_STREAM_URI,
         stream: bool = True,
-        initial_reconnect_delay: float = 1,
+        initial_reconnect_delay: float = DEFAULT_INITIAL_RECONNECT_DELAY,
         defaults: dict = {},
         send_events: Optional[bool] = None,
         update_processor_class: Optional[Callable[['Config', FeatureStore, Event], UpdateProcessor]] = None,
-        poll_interval: float = 30,
+        poll_interval: float = DEFAULT_POLL_INTERVAL,
         use_ldd: bool = False,
         feature_store: Optional[FeatureStore] = None,
         feature_requester_class=None,
@@ -402,7 +407,7 @@ def __init__(
         self.__update_processor_class = update_processor_class
         self.__stream = stream
         self.__initial_reconnect_delay = initial_reconnect_delay
-        self.__poll_interval = max(poll_interval, 30.0)
+        self.__poll_interval = max(poll_interval, DEFAULT_POLL_INTERVAL)
         self.__use_ldd = use_ldd
         self.__feature_store = InMemoryFeatureStore() if not feature_store else feature_store
         self.__event_processor_class = event_processor_class

ldclient/impl/aio/transport.py

@@ -118,11 +118,16 @@ def __init__(self, config, session: Optional[aiohttp.ClientSession] = None, prox
         self._http_options = http_options if http_options is not None else config.http
         self._proxy = proxy if proxy is not None else (self._http_options.http_proxy or None)
 
-    def create(self, url: str, initial_retry_delay: float, query_params=None) -> AsyncSSEClient:
-        """Builds an SSE client for the given stream URL. Headers, timeouts,
-        proxy settings, and the retry/backoff policy come from the SDK config.
-        ``query_params`` is an optional zero-argument callable evaluated on
-        each (re)connect to produce additional query string parameters."""
+    def create(self, url: str, initial_retry_delay: float, query_params=None, sdk_managed_retry: bool = False) -> AsyncSSEClient:
+        """Builds an SSE client for the given stream URL. Headers, timeouts and
+        proxy settings come from the SDK config. ``query_params`` is an
+        optional zero-argument callable evaluated on each (re)connect to
+        produce additional query string parameters.
+
+        ``sdk_managed_retry`` moves the delay between connection attempts to
+        the caller. The SSE client then never waits, and
+        ``initial_retry_delay`` is ignored. When it is false, the SSE client
+        backs off on its own."""
         base_headers = _base_headers(self._config, ASYNC_USER_AGENT)
         aiohttp_request_options: dict = {
             "timeout": aiohttp.ClientTimeout(
@@ -134,6 +139,24 @@ def create(self, url: str, initial_retry_delay: float, query_params=None) -> Asy
         proxy = self._proxy or _get_proxy_url(url)
         if proxy:
             aiohttp_request_options["proxy"] = proxy
+        if sdk_managed_retry:
+            # The SSE client's retry is disabled; the SDK owns the delay. The base
+            # strategy must be passed: omitting it selects the library's backoff.
+            retry_options: dict = {
+                "initial_retry_delay": 0,
+                "retry_delay_strategy": RetryDelayStrategy(),
+                "retry_delay_reset_threshold": 0,
+            }
+        else:
+            retry_options = {
+                "initial_retry_delay": initial_retry_delay,
+                "retry_delay_strategy": RetryDelayStrategy.default(
+                    max_delay=MAX_RETRY_DELAY,
+                    backoff_multiplier=2,
+                    jitter_multiplier=JITTER_RATIO,
+                ),
+                "retry_delay_reset_threshold": BACKOFF_RESET_INTERVAL,
+            }
         return AsyncSSEClient(
             connect=AsyncConnectStrategy.http(
                 url=url,
@@ -143,12 +166,6 @@ def create(self, url: str, initial_retry_delay: float, query_params=None) -> Asy
                 query_params=query_params,
             ),
             error_strategy=ErrorStrategy.always_continue(),  # we'll make error-handling decisions when we see a Fault
-            initial_retry_delay=initial_retry_delay,
-            retry_delay_strategy=RetryDelayStrategy.default(
-                max_delay=MAX_RETRY_DELAY,
-                backoff_multiplier=2,
-                jitter_multiplier=JITTER_RATIO,
-            ),
-            retry_delay_reset_threshold=BACKOFF_RESET_INTERVAL,
             logger=log,
+            **retry_options,
         )

ldclient/impl/datasource/async_polling.py

@@ -9,11 +9,16 @@
 
 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.datasource.datasource_common import async_sink_or_store
+from ldclient.impl.retry import (
+    FailureKind,
+    RetryState,
+    classify_http_status,
+    for_polling
+)
 from ldclient.impl.util import (
     UnsuccessfulResponseException,
-    http_error_message,
-    is_http_error_recoverable,
+    http_error_description,
     log
 )
 from ldclient.interfaces import (
@@ -27,13 +32,22 @@
 
 
 class AsyncPollingUpdateProcessor(AsyncUpdateProcessor):
-    def __init__(self, config: AsyncConfig, requester: AsyncFeatureRequester, store: AsyncFeatureStore, ready: AsyncEvent):
+    """Polls LaunchDarkly for flag data on its own background task.
+
+    The loop reads its wait from the retry state, which ``_fetch_and_store``
+    updates, so a failure can push the next poll further out than the poll
+    interval. See :mod:`ldclient.impl.retry`.
+    """
+
+    def __init__(self, config: AsyncConfig, requester: AsyncFeatureRequester, store: AsyncFeatureStore, ready: AsyncEvent, retry_state: Optional[RetryState] = None):
         self._config = config
         self._data_source_update_sink = config.data_source_update_sink
         self._requester = requester
         self._store = store
         self._ready = ready
-        self._task = AsyncRepeatingTask.at_interval("ldclient.datasource.polling", config.poll_interval, 0, self._fetch_and_store)
+        self._retry = retry_state or for_polling(config.poll_interval)
+        # No initial delay: the first poll is immediate.
+        self._task = AsyncRepeatingTask("ldclient.datasource.polling", self._retry, 0, self._fetch_and_store)
 
     def start(self):
         log.info("Starting AsyncPollingUpdateProcessor with request interval: " + str(self._config.poll_interval))
@@ -43,48 +57,53 @@ def initialized(self):
         return self._ready.is_set() and self._store.initialized
 
     async def stop(self):
-        self.__stop_with_error_info(None)
-        # Wait for the current poll to finish before closing the transport, so we do
-        # not close it while a request is still using it. The close is in a finally
-        # so an owned transport is still released if stop() is cancelled mid-wait.
-        try:
-            await self._task.wait_stopped()
-        finally:
-            await self._requester.close()
-
-    def __stop_with_error_info(self, error: Optional[DataSourceErrorInfo]):
         log.info("Stopping AsyncPollingUpdateProcessor")
         self._task.stop()
 
-        if self._data_source_update_sink is None:
-            return
+        if self._data_source_update_sink is not None:
+            self._data_source_update_sink.update_status(DataSourceState.OFF, None)
 
-        self._data_source_update_sink.update_status(DataSourceState.OFF, error)
+        # Do not close the transport while an in-flight request still uses it.
+        try:
+            await self._task.wait_stopped()
+        finally:
+            await self._requester.close()
 
-    async def _fetch_and_store(self):
+    async def _fetch_and_store(self) -> None:
+        """Makes one poll request and records the outcome on the retry state."""
         try:
             all_data = await self._requester.get_all_data()
-            await sink_or_store(self._data_source_update_sink, self._store).init(all_data)
+            await async_sink_or_store(self._data_source_update_sink, self._store).init(all_data)
+
+            if self._data_source_update_sink is not None:
+                self._data_source_update_sink.update_status(DataSourceState.VALID, None)
+
+            # Report the status before signaling readiness, so a caller that
+            # wakes on readiness cannot still read INITIALIZING.
             if not self._ready.is_set() and self._store.initialized:
                 log.info("AsyncPollingUpdateProcessor initialized ok")
                 self._ready.set()
 
-            if self._data_source_update_sink is not None:
-                self._data_source_update_sink.update_status(DataSourceState.VALID, None)
+            self._retry.record_success()
+            return
         except UnsuccessfulResponseException as e:
+            kind = classify_http_status(e.status)
             error_info = DataSourceErrorInfo(DataSourceErrorKind.ERROR_RESPONSE, e.status, time.time(), str(e))
+            description = "Received %s for polling request" % http_error_description(e.status)
+            level = log.error if kind is FailureKind.UNEXPECTED else log.warning
+            stacktrace = None
+        except Exception as e:
+            kind = FailureKind.NORMAL
+            error_info = DataSourceErrorInfo(DataSourceErrorKind.UNKNOWN, 0, time.time(), str(e))
+            description = "Error encountered when updating flags: %s" % e
+            level = log.error
+            # The exception is passed explicitly: by the time the message is
+            # logged, the handler has exited and exc_info() is empty.
+            stacktrace = e
 
-            http_error_message_result = http_error_message(e.status, "polling request")
-            if not is_http_error_recoverable(e.status):
-                log.error(http_error_message_result)
-                self._ready.set()  # if client is initializing, make it stop waiting; has no effect if already inited
-                self.__stop_with_error_info(error_info)
-            else:
-                log.warning(http_error_message_result)
+        self._retry.record_failure(kind)
+        delay = self._retry.next_delay
+        level("%s - will retry in %.1fs" % (description, delay), exc_info=stacktrace)
 
-                if self._data_source_update_sink is not None:
-                    self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, error_info)
-        except Exception as e:
-            log.exception('Error: Exception encountered when updating flags. %s' % e)
-            if self._data_source_update_sink is not None:
-                self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, DataSourceErrorInfo(DataSourceErrorKind.UNKNOWN, 0, time.time(), str(e)))
+        if self._data_source_update_sink is not None:
+            self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, error_info)

ldclient/impl/datasource/async_status.py

@@ -72,6 +72,11 @@ def update_status(self, new_state: DataSourceState, new_error: Optional[DataSour
 
         old_status = self.__status
 
+        # OFF is terminal. A poll or stream connection that was still in
+        # flight when the data source stopped must not report after it.
+        if old_status.state == DataSourceState.OFF:
+            return
+
         if new_state == DataSourceState.INTERRUPTED and old_status.state == DataSourceState.INITIALIZING:
             new_state = DataSourceState.INITIALIZING