@@ -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
 
@@ -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
 
@@ -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
@@ -253,7 +254,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):
 
@@ -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()
@@ -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."""
 
@@ -257,13 +275,14 @@ def for_streaming(initial_reconnect_delay: float) -> RetryState:
     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.
+    The configured delay is validated here, because ``Config`` reports it as
+    given. It is not clamped to either ceiling: a delay that applies after an
+    unexpected failure may never be shorter than the configured delay, so a
+    configured delay longer than a ceiling raises the 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,
@@ -285,10 +304,14 @@ def for_polling(poll_interval: float) -> RetryState:
     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 validated here, because ``Config`` reports it as
+    given. It is not clamped to a ceiling, unlike the streaming delay. The
+    normal ceiling is the poll interval itself, so a clamp would do nothing,
+    and the extended regime has to keep the cadence even where that is longer
+    than an hour: a wait may never be shorter than the interval the caller
+    configured, 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,
 
@@ -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")
 
 
@@ -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.
     """
 
 
 
@@ -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()
 
@@ -9,6 +9,7 @@
 import json
 import logging
 import ssl
+import time
 from unittest import mock
 
 import aiohttp
@@ -105,9 +106,12 @@ def _retry_state_with(policy: AfterHealthyFor) -> RetryState:
     )
 
 
+ONE_HOUR = 60 * 60
+
+
 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."""
+    """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,
@@ -595,6 +599,37 @@ async def test_a_fresh_stream_starts_a_new_reset_window():
     assert len(set(windows)) == 2, "the second stream reused the first window"
 
 
+@pytest.mark.asyncio
+async def test_an_extended_regime_wait_is_cut_short_by_stop():
+    """Shutdown must not sit through an hour-long backoff. The healthy-stop case
+    is test_stop_closes_sse_and_finishes_task; this one stops mid-wait."""
+    from ld_eventsource.errors import HTTPStatusError
+
+    retry = _fast_retry_state(ONE_HOUR)
+    proc, store, ready, _ = _make_processor(
+        [_start(), _fault(error=HTTPStatusError(401))], retry_state=retry
+    )
+    proc.start()
+    # Confirm the wait under test really is long before measuring the stop.
+    await _wait_until(lambda: retry.next_delay > 60)
+
+    started = time.time()
+    await proc.stop()
+    elapsed = time.time() - started
+
+    assert elapsed < 2, "stop() took %.2fs" % elapsed
+    leaked = [t for t in proc._runner._tasks if not t.done()]
+    assert leaked == [], "stop() returned with the task still running: %r" % leaked
+
+
+@pytest.mark.asyncio
+async def test_stop_before_start_and_stop_twice_are_safe():
+    proc, store, ready, _ = _make_processor([])
+
+    await proc.stop()
+    await proc.stop()
+
+
 @pytest.mark.asyncio
 async def test_stop_closes_sse_and_finishes_task():
     flag = FlagBuilder('f1').version(1).build()