fix: Validate the configured data source intervals in Config

Config accepted any initial_reconnect_delay and clamped poll_interval
with max(), which let NaN and inf through because every comparison
against NaN is false. A NaN interval reaches Event.wait() and the delay
arithmetic downstream. Both are now validated, warn, and fall back to
the documented default; the poll interval keeps its 30s minimum on top.

Move the check to impl/util.py as validate_positive_finite, next to the
validators Config already imports, so config and retry share it without
either importing the other. The retry factories keep their own call: a
RetryState can be built without going through Config.

Also name the delay bounds after the spec's ceiling vocabulary.

Author: jsonbailey

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,
@@ -28,6 +30,7 @@
 from ldclient.impl.util import (
     log,
     validate_application_info,
+    validate_positive_finite,
     validate_sdk_key_format
 )
 from ldclient.interfaces import (
@@ -149,11 +152,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,
@@ -255,8 +258,13 @@ def __init__(
         self.__stream_uri = stream_uri.rstrip('/')
         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.__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.__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,6 +15,7 @@
 from ldclient.impl.util import (
     log,
     validate_application_info,
+    validate_positive_finite,
     validate_sdk_key_format
 )
 from ldclient.interfaces import (
@@ -36,6 +37,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 +301,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,
@@ -401,8 +407,13 @@ def __init__(
         self.__stream_uri = stream_uri.rstrip('/')
         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.__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.__use_ldd = use_ldd
         self.__feature_store = InMemoryFeatureStore() if not feature_store else feature_store
         self.__event_processor_class = event_processor_class

ldclient/impl/retry.py

@@ -18,24 +18,24 @@
 
 # currently excluded from documentation - see docs/README.md
 
-import math
 import random
 import time
 from enum import Enum
 from typing import Optional, Protocol
 
-from ldclient.impl.util import log
+from ldclient.config import (
+    DEFAULT_INITIAL_RECONNECT_DELAY,
+    DEFAULT_POLL_INTERVAL
+)
+from ldclient.impl.util import log, validate_positive_finite
 
-# The documented defaults, in seconds. Each stands in for a configured value
-# that is not a positive, finite number.
-DEFAULT_STREAMING_INITIAL_RECONNECT_DELAY = 1
-DEFAULT_STREAMING_MAX_DELAY = 30
-DEFAULT_POLL_INTERVAL = 30
+# The delay ceiling of the normal regime for streaming, in seconds.
+NORMAL_STREAMING_CEILING_DELAY = 30
 
 # 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
+EXTENDED_CEILING_DELAY = 60 * 60
 
 # How long streaming must operate without a failure before its retry state
 # resets, in seconds.
@@ -152,35 +152,37 @@ class RetryState:
     def __init__(
         self,
         normal_initial_delay: float,
-        normal_ceiling: float,
+        normal_ceiling_delay: float,
         extended_initial_delay: float,
-        extended_ceiling: float,
+        extended_ceiling_delay: float,
         reset_policy: ResetPolicy,
         operating_cadence: float = 0,
     ):
         """
         :param normal_initial_delay: the delay before the first retry in the
             normal regime, in seconds
-        :param normal_ceiling: the longest normal-regime delay, in seconds
+        :param normal_ceiling_delay: 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 extended_ceiling_delay: the longest extended-regime delay, in
+            seconds
         :param reset_policy: decides when the retry state resets
         :param operating_cadence: the wait between healthy operations, in
             seconds; no wait is ever shorter than this. Zero for a component
             that operates continuously.
         """
         self._normal_initial_delay = normal_initial_delay
-        self._normal_ceiling = normal_ceiling
+        self._normal_ceiling_delay = normal_ceiling_delay
         self._extended_initial_delay = extended_initial_delay
-        self._extended_ceiling = extended_ceiling
+        self._extended_ceiling_delay = extended_ceiling_delay
         self._reset_policy = reset_policy
         self._operating_cadence = operating_cadence
 
         self._n = 0
         self._extended = False
         self._min_delay = self._normal_initial_delay
-        self._max_delay = max(self._normal_ceiling, self._normal_initial_delay)
+        self._max_delay = max(self._normal_ceiling_delay, self._normal_initial_delay)
         self._attempts = 0
         # Read before any outcome is recorded, this is the ordinary interval.
         self._next_delay = self._operating_cadence
@@ -212,7 +214,7 @@ def record_failure(self, kind: FailureKind) -> None:
             # extended initial delay.
             self._extended = True
             self._min_delay = self._extended_initial_delay
-            self._max_delay = max(self._extended_ceiling, self._min_delay)
+            self._max_delay = max(self._extended_ceiling_delay, self._min_delay)
             self._n = 1
         else:
             self._n += 1
@@ -245,20 +247,7 @@ def _reset_if_due(self) -> None:
         self._attempts = 0
         self._extended = False
         self._min_delay = self._normal_initial_delay
-        self._max_delay = max(self._normal_ceiling, self._normal_initial_delay)
-
-
-def _positive_finite(value: float, default: float, name: str) -> float:
-    """Returns ``value`` if it is a positive, finite number of seconds, and the
-    default otherwise. A non-finite value would make the jitter arithmetic
-    produce a NaN delay, and a non-positive one would retry with no wait."""
-    if value > 0 and math.isfinite(value):
-        return value
-    log.warning(
-        "%s must be a positive, finite number of seconds; using the default of %ss"
-        % (name, default)
-    )
-    return default
+        self._max_delay = max(self._normal_ceiling_delay, self._normal_initial_delay)
 
 
 def for_streaming(initial_reconnect_delay: float) -> RetryState:
@@ -276,14 +265,14 @@ def for_streaming(initial_reconnect_delay: float) -> RetryState:
 
     The extended regime never starts below the configured delay.
     """
-    initial_reconnect_delay = _positive_finite(
-        initial_reconnect_delay, DEFAULT_STREAMING_INITIAL_RECONNECT_DELAY, 'initial_reconnect_delay'
+    initial_reconnect_delay = validate_positive_finite(
+        initial_reconnect_delay, DEFAULT_INITIAL_RECONNECT_DELAY, 'initial_reconnect_delay', log
     )
     return RetryState(
         normal_initial_delay=initial_reconnect_delay,
-        normal_ceiling=DEFAULT_STREAMING_MAX_DELAY,
+        normal_ceiling_delay=NORMAL_STREAMING_CEILING_DELAY,
         extended_initial_delay=max(EXTENDED_INITIAL_DELAY, initial_reconnect_delay),
-        extended_ceiling=EXTENDED_MAX_DELAY,
+        extended_ceiling_delay=EXTENDED_CEILING_DELAY,
         reset_policy=AfterHealthyFor(STREAMING_RESET_INTERVAL),
         operating_cadence=0
     )
@@ -302,12 +291,12 @@ def for_polling(poll_interval: float) -> RetryState:
     ``Config`` clamps the poll interval, but the documented default stands in
     for anything that reaches here and is not a positive, finite number.
     """
-    poll_interval = _positive_finite(poll_interval, DEFAULT_POLL_INTERVAL, 'poll_interval')
+    poll_interval = validate_positive_finite(poll_interval, DEFAULT_POLL_INTERVAL, 'poll_interval', log)
     return RetryState(
         normal_initial_delay=poll_interval,
-        normal_ceiling=poll_interval,
+        normal_ceiling_delay=poll_interval,
         extended_initial_delay=max(EXTENDED_INITIAL_DELAY, poll_interval),
-        extended_ceiling=EXTENDED_MAX_DELAY,
+        extended_ceiling_delay=EXTENDED_CEILING_DELAY,
         reset_policy=AfterConsecutiveSuccesses(POLLING_RESET_SUCCESSES),
         operating_cadence=poll_interval,
     )

ldclient/impl/util.py

@@ -1,4 +1,5 @@
 import logging
+import math
 import re
 import sys
 import time
@@ -60,6 +61,25 @@ 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.

ldclient/testing/impl/datasource/test_async_polling.py

@@ -63,9 +63,9 @@ def fast_retry_state(delay=0.001):
     real extended-regime delay of five minutes."""
     return RetryState(
         normal_initial_delay=delay,
-        normal_ceiling=delay,
+        normal_ceiling_delay=delay,
         extended_initial_delay=delay,
-        extended_ceiling=delay,
+        extended_ceiling_delay=delay,
         reset_policy=AfterConsecutiveSuccesses(POLLING_RESET_SUCCESSES),
         operating_cadence=delay,
     )

ldclient/testing/impl/datasource/test_async_streaming.py

@@ -22,9 +22,9 @@
 )
 from ldclient.impl.model import ModelEntity
 from ldclient.impl.retry import (
-    DEFAULT_STREAMING_MAX_DELAY,
+    EXTENDED_CEILING_DELAY,
     EXTENDED_INITIAL_DELAY,
-    EXTENDED_MAX_DELAY,
+    NORMAL_STREAMING_CEILING_DELAY,
     STREAMING_RESET_INTERVAL,
     AfterHealthyFor,
     RetryState,
@@ -98,9 +98,9 @@ def _retry_state_with(policy: AfterHealthyFor) -> RetryState:
     test can watch the window."""
     return RetryState(
         normal_initial_delay=0.001,
-        normal_ceiling=0.001,
+        normal_ceiling_delay=0.001,
         extended_initial_delay=0.001,
-        extended_ceiling=0.001,
+        extended_ceiling_delay=0.001,
         reset_policy=policy,
     )
 
@@ -110,9 +110,9 @@ def _fast_retry_state(delay: float = 0.001) -> RetryState:
     real extended-regime delay of five minutes."""
     return RetryState(
         normal_initial_delay=delay,
-        normal_ceiling=delay,
+        normal_ceiling_delay=delay,
         extended_initial_delay=delay,
-        extended_ceiling=delay,
+        extended_ceiling_delay=delay,
         reset_policy=AfterHealthyFor(STREAMING_RESET_INTERVAL),
     )
 
@@ -137,9 +137,9 @@ def _zero_delay_retry_state() -> RetryState:
     real, so a misclassification still shows up in ``max_delay``."""
     return RetryState(
         normal_initial_delay=0,
-        normal_ceiling=DEFAULT_STREAMING_MAX_DELAY,
+        normal_ceiling_delay=NORMAL_STREAMING_CEILING_DELAY,
         extended_initial_delay=EXTENDED_INITIAL_DELAY,
-        extended_ceiling=EXTENDED_MAX_DELAY,
+        extended_ceiling_delay=EXTENDED_CEILING_DELAY,
         reset_policy=AfterHealthyFor(STREAMING_RESET_INTERVAL),
     )
 
@@ -488,7 +488,7 @@ async def test_transport_failures_stay_in_the_normal_regime(error):
     assert await asyncio.wait_for(proc._handle_error(error), timeout=2.0)
 
     assert not retry._extended
-    assert retry._max_delay == DEFAULT_STREAMING_MAX_DELAY
+    assert retry._max_delay == NORMAL_STREAMING_CEILING_DELAY
 
 
 class _NoSleep:

ldclient/testing/impl/datasource/test_polling_processor.py

@@ -51,9 +51,9 @@ def fast_retry_state(delay=0.05):
     real extended-regime delay of five minutes."""
     return RetryState(
         normal_initial_delay=delay,
-        normal_ceiling=delay,
+        normal_ceiling_delay=delay,
         extended_initial_delay=delay,
-        extended_ceiling=delay,
+        extended_ceiling_delay=delay,
         reset_policy=AfterConsecutiveSuccesses(POLLING_RESET_SUCCESSES),
         operating_cadence=delay,
     )

ldclient/testing/impl/datasource/test_streaming.py

@@ -64,9 +64,9 @@ def fast_retry_state(delay=brief_delay):
     real extended-regime delay of five minutes."""
     return RetryState(
         normal_initial_delay=delay,
-        normal_ceiling=delay,
+        normal_ceiling_delay=delay,
         extended_initial_delay=delay,
-        extended_ceiling=delay,
+        extended_ceiling_delay=delay,
         reset_policy=AfterHealthyFor(STREAMING_RESET_INTERVAL),
     )
 
@@ -565,9 +565,9 @@ def test_several_messages_on_one_stream_do_not_extend_the_reset_window():
             policy = AfterHealthyFor(STREAMING_RESET_INTERVAL)
             retry = RetryState(
                 normal_initial_delay=brief_delay,
-                normal_ceiling=brief_delay,
+                normal_ceiling_delay=brief_delay,
                 extended_initial_delay=brief_delay,
-                extended_ceiling=brief_delay,
+                extended_ceiling_delay=brief_delay,
                 reset_policy=policy,
             )
             # The clock moves on every read, so a window that had been
@@ -601,9 +601,9 @@ def test_a_fresh_stream_starts_a_new_reset_window():
                 policy = AfterHealthyFor(STREAMING_RESET_INTERVAL)
                 retry = RetryState(
                     normal_initial_delay=brief_delay,
-                    normal_ceiling=brief_delay,
+                    normal_ceiling_delay=brief_delay,
                     extended_initial_delay=brief_delay,
-                    extended_ceiling=brief_delay,
+                    extended_ceiling_delay=brief_delay,
                     reset_policy=policy,
                 )
                 windows = record_healthy_windows(policy)