fix: Validate data source intervals in the retry state and bound the wait

Author: jsonbailey

ldclient/async_config.py

@@ -30,7 +30,6 @@
 from ldclient.impl.util import (
     log,
     validate_application_info,
-    validate_positive_finite,
     validate_sdk_key_format
 )
 from ldclient.interfaces import (
@@ -258,13 +257,8 @@ def __init__(
         self.__stream_uri = stream_uri.rstrip('/')
         self.__update_processor_class = update_processor_class
         self.__stream = stream
-        self.__initial_reconnect_delay = validate_positive_finite(
-            initial_reconnect_delay, DEFAULT_INITIAL_RECONNECT_DELAY, 'initial_reconnect_delay', log
-        )
-        self.__poll_interval = max(
-            validate_positive_finite(poll_interval, DEFAULT_POLL_INTERVAL, 'poll_interval', log),
-            DEFAULT_POLL_INTERVAL,
-        )
+        self.__initial_reconnect_delay = initial_reconnect_delay
+        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/config.py

@@ -15,7 +15,6 @@
 from ldclient.impl.util import (
     log,
     validate_application_info,
-    validate_positive_finite,
     validate_sdk_key_format
 )
 from ldclient.interfaces import (
@@ -407,13 +406,8 @@ def __init__(
         self.__stream_uri = stream_uri.rstrip('/')
         self.__update_processor_class = update_processor_class
         self.__stream = stream
-        self.__initial_reconnect_delay = validate_positive_finite(
-            initial_reconnect_delay, DEFAULT_INITIAL_RECONNECT_DELAY, 'initial_reconnect_delay', log
-        )
-        self.__poll_interval = max(
-            validate_positive_finite(poll_interval, DEFAULT_POLL_INTERVAL, 'poll_interval', log),
-            DEFAULT_POLL_INTERVAL,
-        )
+        self.__initial_reconnect_delay = initial_reconnect_delay
+        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

@@ -140,8 +140,7 @@ def create(self, url: str, initial_retry_delay: float, query_params=None, sdk_ma
         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.
+            # The SSE client's retry is disabled; the SDK owns the delay.
             retry_options: dict = {
                 "initial_retry_delay": 0,
                 "retry_delay_strategy": RetryDelayStrategy(),

ldclient/impl/datasource/async_streaming.py

@@ -126,11 +126,8 @@ async def _run(self):
                             log.info("AsyncStreamingUpdateProcessor initialized ok.")
                             self._ready.set()
                 elif isinstance(action, Fault):
-                    # 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.
+                    # A Fault with no error is a clean close. An interrupt the
+                    # SDK asked for is not a failure.
                     if action.error is None:
                         if self._interrupted_by_sdk:
                             self._interrupted_by_sdk = False

ldclient/impl/datasource/streaming.py

@@ -1,5 +1,6 @@
 import json
 import time
+from threading import TIMEOUT_MAX
 from threading import Event as ThreadEvent
 from threading import Thread
 from typing import Callable, Optional
@@ -107,11 +108,8 @@ def run(self):
                         log.info("StreamingUpdateProcessor initialized ok.")
                         self._ready.set()
             elif isinstance(action, Fault):
-                # 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.
+                # A Fault with no error is a clean close. An interrupt the SDK
+                # asked for is not a failure.
                 if action.error is None:
                     if self._interrupted_by_sdk:
                         self._interrupted_by_sdk = False
@@ -139,13 +137,7 @@ 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
-            # 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.
+            # The SSE client's retry is disabled; the SDK owns the delay.
             initial_retry_delay=0,
             retry_delay_strategy=RetryDelayStrategy(),
             retry_delay_reset_threshold=0,
@@ -253,7 +245,7 @@ def _handle_error(self, error: Exception) -> bool:
             self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, error_info)
 
         self._connection_attempt_start_time = time.time() + delay
-        return not self._stop_event.wait(delay)
+        return not self._stop_event.wait(min(delay, TIMEOUT_MAX))
 
     # magic methods for "with" statement (used in testing)
     def __enter__(self):

ldclient/impl/repeating_task.py

@@ -1,4 +1,4 @@
-from threading import Event, Thread
+from threading import TIMEOUT_MAX, Event, Thread
 from typing import Any, Callable
 
 from ldclient.impl.delay import DelaySource, FixedDelay
@@ -66,7 +66,7 @@ def stop(self):
 
     def _run(self):
         if self.__initial_delay > 0:
-            if self.__stop.wait(self.__initial_delay):
+            if self.__stop.wait(min(self.__initial_delay, TIMEOUT_MAX)):
                 return
         stopped = self.__stop.is_set()
         while not stopped:
@@ -77,4 +77,4 @@ def _run(self):
             # The wait starts when the callback returns, so a slow callback
             # never shortens it.
             delay = self.__delays.next_delay
-            stopped = self.__stop.wait(delay) if delay > 0 else self.__stop.is_set()
+            stopped = self.__stop.wait(min(delay, TIMEOUT_MAX)) if delay > 0 else self.__stop.is_set()

ldclient/impl/retry.py

@@ -18,6 +18,7 @@
 
 # currently excluded from documentation - see docs/README.md
 
+import math
 import random
 import time
 from enum import Enum
@@ -27,7 +28,7 @@
     DEFAULT_INITIAL_RECONNECT_DELAY,
     DEFAULT_POLL_INTERVAL
 )
-from ldclient.impl.util import log, validate_positive_finite
+from ldclient.impl.util import log
 
 # The delay ceiling of the normal regime for streaming, in seconds.
 NORMAL_STREAMING_CEILING_DELAY = 30
@@ -53,6 +54,23 @@
 _MAX_BACKOFF_EXPONENT = 30
 
 
+def _usable_delay(value: float, default: float, name: str, ceiling: float = math.inf) -> float:
+    """
+    Returns the delay to use, clamped to the ceiling. A value that is
+    not a positive, finite number of seconds is replaced by the default.
+
+    :param value: the configured number of seconds
+    :param default: the value to use when ``value`` is not usable
+    :param name: the option name, for the warning message
+    :param ceiling: the longest delay allowed
+    """
+
+    if value > 0 and math.isfinite(value):
+        return min(value, ceiling)
+    log.warning("%s must be a positive, finite number of seconds; using the default of %ss" % (name, default))
+    return default
+
+
 class FailureKind(Enum):
     """How a failure is classified, which decides how long the next wait is."""
 
@@ -251,19 +269,12 @@ def for_streaming(initial_reconnect_delay: float) -> RetryState:
     """
     Builds the retry state for a streaming data source.
 
-    Streaming's operating cadence is zero, so there is no delay during
-    healthy operation. Stream failures use either the normal or extended
-    initial delay to determine their backoff wait. A stream returns to
-    healthy operation after establishing a successful connection with no
-    failures during the ``STREAMING_RESET_INTERVAL``.
-
-    ``Config`` validates the configured delay, so this guard only catches a
-    state built without it.
-
-    The extended regime never starts below the configured delay.
+    Streaming's cadence is zero, so a healthy stream never waits. An invalid
+    delay value is replaced by the documented default; one longer than a
+    ceiling raises that bound rather than being cut down to it.
     """
-    initial_reconnect_delay = validate_positive_finite(
-        initial_reconnect_delay, DEFAULT_INITIAL_RECONNECT_DELAY, 'initial_reconnect_delay', log
+    initial_reconnect_delay = _usable_delay(
+        initial_reconnect_delay, DEFAULT_INITIAL_RECONNECT_DELAY, 'initial_reconnect_delay'
     )
     return RetryState(
         normal_initial_delay=initial_reconnect_delay,
@@ -279,16 +290,12 @@ def for_polling(poll_interval: float) -> RetryState:
     """
     Builds the retry state for a polling data source.
 
-    The poll interval is polling's operating cadence, so no wait is ever
-    shorter than it. In the normal regime the delay bounds are the poll
-    interval itself, which means a normal failure simply polls again on
-    schedule. Polling is healthy on any successful poll, and resets after two
-    in a row.
-
-    ``Config`` validates and clamps the poll interval, so this guard only
-    catches a state built without it.
+    The poll interval is polling's cadence and its normal ceiling, so a normal
+    failure waits the interval rather than backing off past it. An invalid
+    interval is replaced by the documented default. No wait is ever shorter
+    than the interval, so the cadence wins over the extended ceiling.
     """
-    poll_interval = validate_positive_finite(poll_interval, DEFAULT_POLL_INTERVAL, 'poll_interval', log)
+    poll_interval = _usable_delay(poll_interval, DEFAULT_POLL_INTERVAL, 'poll_interval')
     return RetryState(
         normal_initial_delay=poll_interval,
         normal_ceiling_delay=poll_interval,

ldclient/impl/util.py

@@ -1,5 +1,4 @@
 import logging
-import math
 import re
 import sys
 import time
@@ -61,25 +60,6 @@ def validate_application_value(value: Any, name: str, logger: logging.Logger) ->
     return value
 
 
-def validate_positive_finite(value: float, default: float, name: str, logger: logging.Logger) -> float:
-    """
-    Validates that a number of seconds is positive and finite.
-
-    A non-finite value makes later arithmetic produce NaN, and a non-positive
-    one makes a wait no wait at all.
-
-    :param value: the number of seconds to validate
-    :param default: the value to use when ``value`` is not usable
-    :param name: the option name, for the warning message
-    :param logger: the logger to use for logging warnings
-    :return: ``value``, or ``default`` if ``value`` is not positive and finite
-    """
-    if value > 0 and math.isfinite(value):
-        return value
-    logger.warning("%s must be a positive, finite number of seconds; using the default of %ss" % (name, default))
-    return default
-
-
 def validate_sdk_key_format(sdk_key: str, logger: logging.Logger) -> str:
     """
     Validates that an SDK key does not contain invalid characters and is not too long for our systems.
@@ -155,14 +135,11 @@ 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 treated as fatal
-    return True
+        return status in _RETRYABLE_STATUSES  # all other 4xx besides these are unrecoverable
+    return True  # all other errors are recoverable
 
 
 def http_error_description(status):
@@ -171,11 +148,8 @@ 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.
+    Deprecated. The FDv1 data sources build their own message, so that it can
+    report the real retry delay.
     """
     return "Received %s for %s - %s" % (http_error_description(status), context, retryable_message if is_http_error_recoverable(status) else "giving up permanently")
 

ldclient/interfaces.py

@@ -1013,7 +1013,8 @@ class DataSourceState(Enum):
     """
     Indicates that the data source has been permanently shut down.
 
-    This means the SDK client was explicitly shut down, or that its configuration could not be parsed.
+    This could be because the SDK client was explicitly shut down, because its configuration could not
+    be parsed, or because the data source encountered a condition it will not retry.
     """
 
 

ldclient/testing/impl/datasource/test_async_polling.py

@@ -5,6 +5,7 @@
 import asyncio
 import logging
 import ssl
+import time
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import aiohttp
@@ -58,9 +59,12 @@ def make_config(**kwargs):
 )
 
 
+ONE_HOUR = 60 * 60
+
+
 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."""
+    """A retry state whose every delay is ``delay``: small enough to skip the
+    real extended-regime wait, or large enough to prove a stop interrupts one."""
     return RetryState(
         normal_initial_delay=delay,
         normal_ceiling_delay=delay,
@@ -469,6 +473,30 @@ async def close():
 
         assert order == ['poll_done', 'transport_closed']
 
+    @pytest.mark.asyncio
+    async def test_an_extended_regime_wait_is_cut_short_by_stop(self):
+        """Shutdown must not sit through an hour-long backoff. The in-flight-poll
+        case is test_stop_cancels_polling_task_cleanly; this one stops while the
+        task is waiting between polls."""
+        retry = fast_retry_state(ONE_HOUR)
+        processor = make_processor(retry_state=retry)
+        processor._requester.get_all_data = AsyncMock(
+            side_effect=UnsuccessfulResponseException(401)
+        )
+
+        processor.start()
+        # Confirm the wait under test really is long before measuring the stop.
+        deadline = time.time() + 2
+        while retry.next_delay <= 60 and time.time() < deadline:
+            await asyncio.sleep(0.01)
+        assert retry.next_delay > 60, "the wait under test should be minutes long"
+
+        started = time.time()
+        await processor.stop()
+        elapsed = time.time() - started
+
+        assert elapsed < 2, "stop() took %.2fs" % elapsed
+
     @pytest.mark.asyncio
     async def test_stop_cancels_polling_task_cleanly(self):
         store = MockAsyncFeatureStore()