feat: Remove permanent failure modes from FDv1 following RETRY spec

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

@@ -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.
         """

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

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,33 +189,41 @@ 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")
         except AttributeError:
             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:

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

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

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

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)