fix: Prevent long reconnect delays or skipped backoff after a system clock change (#80)

Author: jsonbailey

ld_eventsource/async_client.py

@@ -185,7 +185,7 @@ def next_retry_delay(self) -> float:
 
     def _compute_next_retry_delay(self):
         if self.__retry_delay_reset_threshold > 0 and self._retry_reset_baseline != 0:
-            now = time.time()
+            now = time.monotonic()
             connection_duration = now - self._retry_reset_baseline
             if connection_duration >= self.__retry_delay_reset_threshold:
                 self.__current_retry_delay_strategy = self.__base_retry_delay_strategy
@@ -203,7 +203,7 @@ async def _try_start(self, can_return_fault: bool):
                     self.__next_retry_delay
                     if self.__disconnected_time == 0
                     else self.__next_retry_delay
-                    - (time.time() - self.__disconnected_time)
+                    - (time.monotonic() - self.__disconnected_time)
                 )
                 if delay > 0:
                     self.__logger.info("Will reconnect after delay of %fs" % delay)
@@ -213,7 +213,7 @@ async def _try_start(self, can_return_fault: bool):
                     self.__last_event_id
                 )
             except Exception as e:
-                self.__disconnected_time = time.time()
+                self.__disconnected_time = time.monotonic()
                 self._compute_next_retry_delay()
                 fail_or_continue, self.__current_error_strategy = (
                     self.__current_error_strategy.apply(e)
@@ -223,7 +223,7 @@ async def _try_start(self, can_return_fault: bool):
                 if can_return_fault:
                     return Fault(e)
                 continue
-            self._retry_reset_baseline = time.time()
+            self._retry_reset_baseline = time.monotonic()
             self.__current_error_strategy = self.__base_error_strategy
             self.__interrupted = False
             return Start(self.__connection_result.headers)

ld_eventsource/config/error_strategy.py

@@ -133,9 +133,9 @@ def apply(self, exception: Optional[Exception]) -> Tuple[bool, ErrorStrategy]:
         if self.__start_time == 0:
             return (
                 ErrorStrategy.CONTINUE,
-                _TimeLimitErrorStrategy(self.__max_time, time.time()),
+                _TimeLimitErrorStrategy(self.__max_time, time.monotonic()),
             )
-        if (time.time() - self.__start_time) < self.__max_time:
+        if (time.monotonic() - self.__start_time) < self.__max_time:
             return (ErrorStrategy.CONTINUE, self)
         return (ErrorStrategy.FAIL, self)
 

ld_eventsource/sse_client.py

@@ -276,7 +276,7 @@ def _compute_next_retry_delay(self):
         # it should continue to double until the retry maximum, and then hold
         # steady (- jitter).
         if self.__retry_delay_reset_threshold > 0 and self._retry_reset_baseline != 0:
-            now = time.time()
+            now = time.monotonic()
             connection_duration = now - self._retry_reset_baseline
             if connection_duration >= self.__retry_delay_reset_threshold:
                 self.__current_retry_delay_strategy = self.__base_retry_delay_strategy
@@ -294,7 +294,7 @@ def _try_start(self, can_return_fault: bool) -> Union[None, Start, Fault]:
                     self.__next_retry_delay
                     if self.__disconnected_time == 0
                     else self.__next_retry_delay
-                    - (time.time() - self.__disconnected_time)
+                    - (time.monotonic() - self.__disconnected_time)
                 )
                 if delay > 0:
                     self.__logger.info("Will reconnect after delay of %fs" % delay)
@@ -304,7 +304,7 @@ def _try_start(self, can_return_fault: bool) -> Union[None, Start, Fault]:
                     self.__last_event_id
                 )
             except Exception as e:
-                self.__disconnected_time = time.time()
+                self.__disconnected_time = time.monotonic()
                 self._compute_next_retry_delay()
                 fail_or_continue, self.__current_error_strategy = (
                     self.__current_error_strategy.apply(e)
@@ -316,7 +316,7 @@ def _try_start(self, can_return_fault: bool) -> Union[None, Start, Fault]:
                 # If can_return_fault is false, it means the caller explicitly called start(), in
                 # which case there's no way to return a Fault so we just keep retrying transparently.
                 continue
-            self._retry_reset_baseline = time.time()
+            self._retry_reset_baseline = time.monotonic()
             self.__current_error_strategy = self.__base_error_strategy
             self.__interrupted = False
             return Start(self.__connection_result.headers)

ld_eventsource/testing/test_async_sse_client_retry.py

@@ -1,4 +1,6 @@
 import asyncio
+import itertools
+from unittest.mock import patch
 
 import pytest
 
@@ -187,7 +189,8 @@ async def test_retry_delay_gets_reset_after_threshold():
         assert isinstance(item4, Fault)
         assert client.next_retry_delay == initial_delay * 2
 
-        await asyncio.sleep(retry_delay_reset_threshold)
+        # Overshoot the threshold, because the clock can tick in ~15ms steps on Windows.
+        await asyncio.sleep(retry_delay_reset_threshold * 1.5)
 
         item5 = await all_iter.__anext__()
         assert isinstance(item5, Fault)
@@ -198,3 +201,71 @@ async def test_retry_delay_gets_reset_after_threshold():
         item6 = await all_iter.__anext__()
         assert isinstance(item6, Fault)
         assert client.next_retry_delay == initial_delay * 2
+
+
+@pytest.mark.asyncio
+async def test_reconnect_delay_ignores_wall_clock_step():
+    initial_delay = 0.05
+    mock = MockAsyncConnectStrategy(
+        AsyncRejectConnection(HTTPStatusError(503)),
+        AsyncRespondWithData("data: data1\n\n"),
+        AsyncExpectNoMoreRequests(),
+    )
+    real_sleep = asyncio.sleep
+    delays = []
+
+    async def fake_sleep(delay, *args, **kwargs):
+        delays.append(delay)
+        await real_sleep(0)
+
+    # Each wall-clock read steps back one hour.
+    wall_clock = itertools.count(1_000_000_000, -3600)
+    with patch('time.time', side_effect=lambda: next(wall_clock)), patch(
+        'ld_eventsource.async_client.asyncio.sleep', fake_sleep
+    ):
+        async with AsyncSSEClient(
+            connect=mock,
+            error_strategy=retry_for_status(503),
+            initial_retry_delay=initial_delay,
+            retry_delay_strategy=RetryDelayStrategy.default(jitter_multiplier=None),
+        ) as client:
+            await client.start()
+
+    assert len(delays) == 1
+    assert 0 < delays[0] <= initial_delay
+
+
+@pytest.mark.asyncio
+async def test_retry_delay_reset_ignores_wall_clock_step():
+    initial_delay = 0.005
+    mock = MockAsyncConnectStrategy(
+        AsyncRespondWithData("data: data1\n\n"),
+        AsyncRejectConnection(HTTPStatusError(503)),
+    )
+    real_sleep = asyncio.sleep
+
+    async def fake_sleep(delay, *args, **kwargs):
+        await real_sleep(0)
+
+    # Each wall-clock read steps forward one hour.
+    wall_clock = itertools.count(1_000_000_000, 3600)
+    with patch('time.time', side_effect=lambda: next(wall_clock)), patch(
+        'ld_eventsource.async_client.asyncio.sleep', fake_sleep
+    ):
+        async with AsyncSSEClient(
+            connect=mock,
+            error_strategy=ErrorStrategy.always_continue(),
+            initial_retry_delay=initial_delay,
+            retry_delay_reset_threshold=1000,
+            retry_delay_strategy=RetryDelayStrategy.default(jitter_multiplier=None),
+        ) as client:
+            all_iter = client.all.__aiter__()
+            assert isinstance(await all_iter.__anext__(), Start)
+            assert isinstance(await all_iter.__anext__(), Event)
+            assert isinstance(await all_iter.__anext__(), Fault)
+            assert client.next_retry_delay == initial_delay
+
+            # The connection lasted far less than the threshold, so the
+            # backoff must not reset.
+            assert isinstance(await all_iter.__anext__(), Fault)
+            assert client.next_retry_delay == initial_delay * 2

ld_eventsource/testing/test_error_strategy.py

@@ -1,4 +1,6 @@
+import itertools
 import time
+from unittest.mock import patch
 
 from ld_eventsource.config import *
 
@@ -35,15 +37,34 @@ def test_max_attempts():
 
 
 def test_max_time():
-    strategy = ErrorStrategy.continue_with_time_limit(0.2)
+    max_time = 0.2
+    strategy = ErrorStrategy.continue_with_time_limit(max_time)
 
     should_raise, next1 = strategy.apply(err)
     assert should_raise is False
 
     should_raise, next2 = next1.apply(err)
     assert should_raise is False
 
-    time.sleep(0.2)
+    # Overshoot the limit, because the clock can tick in ~15ms steps on Windows.
+    time.sleep(max_time * 1.5)
 
     should_raise, next3 = next2.apply(err)
     assert should_raise is True
+
+
+def test_max_time_ignores_wall_clock_step():
+    max_time = 0.2
+    strategy = ErrorStrategy.continue_with_time_limit(max_time)
+
+    # Each wall-clock read steps back one hour.
+    wall_clock = itertools.count(1_000_000_000, -3600)
+    with patch('time.time', side_effect=lambda: next(wall_clock)):
+        should_raise, next1 = strategy.apply(err)
+        assert should_raise is False
+
+        # Overshoot the limit, because the clock can tick in ~15ms steps on Windows.
+        time.sleep(max_time * 1.5)
+
+        should_raise, next2 = next1.apply(err)
+        assert should_raise is True

ld_eventsource/testing/test_sse_client_retry.py

@@ -1,4 +1,6 @@
+import itertools
 from time import sleep
+from unittest.mock import patch
 
 from ld_eventsource import *
 from ld_eventsource.actions import *
@@ -138,8 +140,8 @@ def test_retry_delay_gets_reset_after_threshold():
         assert isinstance(item4, Fault)
         assert client.next_retry_delay == initial_delay * 2
 
-        # Sleeping the threshold should reset the retry thresholds
-        sleep(retry_delay_reset_threshold)
+        # Overshoot the threshold, because the clock can tick in ~15ms steps on Windows.
+        sleep(retry_delay_reset_threshold * 1.5)
 
         # Which we see it does here
         item5 = next(all)
@@ -154,6 +156,59 @@ def test_retry_delay_gets_reset_after_threshold():
         assert client.next_retry_delay == initial_delay * 2
 
 
+def test_retry_delay_reset_ignores_wall_clock_step():
+    initial_delay = 0.005
+    mock = MockConnectStrategy(
+        RespondWithData("data: data1\n\n"),
+        RejectConnection(HTTPStatusError(503)),
+    )
+    # Each wall-clock read steps forward one hour.
+    wall_clock = itertools.count(1_000_000_000, 3600)
+    with patch('time.time', side_effect=lambda: next(wall_clock)), patch('time.sleep'):
+        with SSEClient(
+            connect=mock,
+            error_strategy=ErrorStrategy.always_continue(),
+            initial_retry_delay=initial_delay,
+            retry_delay_reset_threshold=1000,
+            retry_delay_strategy=RetryDelayStrategy.default(jitter_multiplier=None),
+        ) as client:
+            all = client.all
+            assert isinstance(next(all), Start)
+            assert isinstance(next(all), Event)
+            assert isinstance(next(all), Fault)
+            assert client.next_retry_delay == initial_delay
+
+            # The connection lasted far less than the threshold, so the
+            # backoff must not reset.
+            assert isinstance(next(all), Fault)
+            assert client.next_retry_delay == initial_delay * 2
+
+
+def test_reconnect_delay_ignores_wall_clock_step():
+    initial_delay = 0.05
+    mock = MockConnectStrategy(
+        RejectConnection(HTTPStatusError(503)),
+        RespondWithData("data: data1\n\n"),
+        ExpectNoMoreRequests(),
+    )
+    # Each wall-clock read steps back one hour.
+    wall_clock = itertools.count(1_000_000_000, -3600)
+    with patch('time.time', side_effect=lambda: next(wall_clock)), patch(
+        'time.sleep'
+    ) as fake_sleep:
+        with SSEClient(
+            connect=mock,
+            error_strategy=retry_for_status(503),
+            initial_retry_delay=initial_delay,
+            retry_delay_strategy=RetryDelayStrategy.default(jitter_multiplier=None),
+        ) as client:
+            client.start()
+
+    delays = [c.args[0] for c in fake_sleep.call_args_list]
+    assert len(delays) == 1
+    assert 0 < delays[0] <= initial_delay
+
+
 def test_old_connection_closed_on_fault_reconnect():
     # When a stream ends and the client reconnects, the old connection must be closed
     # (rather than left for garbage collection) before the new one is opened.