From 8e17d5fb5f1dea29c2ea2f805802e89b21bb0df7 Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Mon, 19 Oct 2026 09:15:18 -0500 Subject: [PATCH 1/4] fix: Keep reconnect delays correct when the system clock changes --- ld_eventsource/async_client.py | 8 ++--- ld_eventsource/config/error_strategy.py | 4 +-- ld_eventsource/sse_client.py | 8 ++--- .../testing/test_async_sse_client_retry.py | 34 +++++++++++++++++++ ld_eventsource/testing/test_error_strategy.py | 17 ++++++++++ .../testing/test_sse_client_retry.py | 27 +++++++++++++++ 6 files changed, 88 insertions(+), 10 deletions(-) diff --git a/ld_eventsource/async_client.py b/ld_eventsource/async_client.py index d7b3f7f..3e69a7d 100644 --- a/ld_eventsource/async_client.py +++ b/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) diff --git a/ld_eventsource/config/error_strategy.py b/ld_eventsource/config/error_strategy.py index 2014d86..85fa00e 100644 --- a/ld_eventsource/config/error_strategy.py +++ b/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) diff --git a/ld_eventsource/sse_client.py b/ld_eventsource/sse_client.py index da61d4f..2cc0198 100644 --- a/ld_eventsource/sse_client.py +++ b/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) diff --git a/ld_eventsource/testing/test_async_sse_client_retry.py b/ld_eventsource/testing/test_async_sse_client_retry.py index 8fc08d4..af3fbe2 100644 --- a/ld_eventsource/testing/test_async_sse_client_retry.py +++ b/ld_eventsource/testing/test_async_sse_client_retry.py @@ -1,4 +1,6 @@ import asyncio +import itertools +from unittest.mock import patch import pytest @@ -198,3 +200,35 @@ 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 diff --git a/ld_eventsource/testing/test_error_strategy.py b/ld_eventsource/testing/test_error_strategy.py index ed21b2a..d769981 100644 --- a/ld_eventsource/testing/test_error_strategy.py +++ b/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 * @@ -47,3 +49,18 @@ def test_max_time(): should_raise, next3 = next2.apply(err) assert should_raise is True + + +def test_max_time_ignores_wall_clock_step(): + strategy = ErrorStrategy.continue_with_time_limit(0.2) + + # 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 + + time.sleep(0.2) + + should_raise, next2 = next1.apply(err) + assert should_raise is True diff --git a/ld_eventsource/testing/test_sse_client_retry.py b/ld_eventsource/testing/test_sse_client_retry.py index 762cee9..f5d2c69 100644 --- a/ld_eventsource/testing/test_sse_client_retry.py +++ b/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 * @@ -154,6 +156,31 @@ def test_retry_delay_gets_reset_after_threshold(): 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. From 04b0a37fe7164ebe1c357fb2907a5f48967489be Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Mon, 19 Oct 2026 09:18:06 -0500 Subject: [PATCH 2/4] test: Sleep past the time limit so coarse Windows clock ticks do not flake --- ld_eventsource/testing/test_error_strategy.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/ld_eventsource/testing/test_error_strategy.py b/ld_eventsource/testing/test_error_strategy.py index d769981..178c961 100644 --- a/ld_eventsource/testing/test_error_strategy.py +++ b/ld_eventsource/testing/test_error_strategy.py @@ -45,7 +45,7 @@ def test_max_time(): should_raise, next2 = next1.apply(err) assert should_raise is False - time.sleep(0.2) + time.sleep(0.25) should_raise, next3 = next2.apply(err) assert should_raise is True @@ -60,7 +60,7 @@ def test_max_time_ignores_wall_clock_step(): should_raise, next1 = strategy.apply(err) assert should_raise is False - time.sleep(0.2) + time.sleep(0.25) should_raise, next2 = next1.apply(err) assert should_raise is True From 74c6ccfe5c48eab8e89d4ca8a80a1a8badea425b Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Mon, 19 Oct 2026 09:32:39 -0500 Subject: [PATCH 3/4] test: Cover the backoff-reset window under a wall-clock step --- .../testing/test_async_sse_client_retry.py | 38 ++++++++++++++++++- .../testing/test_sse_client_retry.py | 30 ++++++++++++++- 2 files changed, 66 insertions(+), 2 deletions(-) diff --git a/ld_eventsource/testing/test_async_sse_client_retry.py b/ld_eventsource/testing/test_async_sse_client_retry.py index af3fbe2..4eb1d01 100644 --- a/ld_eventsource/testing/test_async_sse_client_retry.py +++ b/ld_eventsource/testing/test_async_sse_client_retry.py @@ -189,7 +189,7 @@ 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) + await asyncio.sleep(retry_delay_reset_threshold + 0.05) item5 = await all_iter.__anext__() assert isinstance(item5, Fault) @@ -232,3 +232,39 @@ async def fake_sleep(delay, *args, **kwargs): 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 diff --git a/ld_eventsource/testing/test_sse_client_retry.py b/ld_eventsource/testing/test_sse_client_retry.py index f5d2c69..d9f38d8 100644 --- a/ld_eventsource/testing/test_sse_client_retry.py +++ b/ld_eventsource/testing/test_sse_client_retry.py @@ -141,7 +141,7 @@ def test_retry_delay_gets_reset_after_threshold(): assert client.next_retry_delay == initial_delay * 2 # Sleeping the threshold should reset the retry thresholds - sleep(retry_delay_reset_threshold) + sleep(retry_delay_reset_threshold + 0.05) # Which we see it does here item5 = next(all) @@ -156,6 +156,34 @@ 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( From b52853c1d860ef04c4929d4edae5eab8e37acd45 Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Mon, 19 Oct 2026 09:58:47 -0500 Subject: [PATCH 4/4] test: Overshoot timing thresholds by a ratio instead of a fixed offset --- .../testing/test_async_sse_client_retry.py | 3 ++- ld_eventsource/testing/test_error_strategy.py | 12 ++++++++---- ld_eventsource/testing/test_sse_client_retry.py | 4 ++-- 3 files changed, 12 insertions(+), 7 deletions(-) diff --git a/ld_eventsource/testing/test_async_sse_client_retry.py b/ld_eventsource/testing/test_async_sse_client_retry.py index 4eb1d01..4c21a3b 100644 --- a/ld_eventsource/testing/test_async_sse_client_retry.py +++ b/ld_eventsource/testing/test_async_sse_client_retry.py @@ -189,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 + 0.05) + # 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) diff --git a/ld_eventsource/testing/test_error_strategy.py b/ld_eventsource/testing/test_error_strategy.py index 178c961..336c96c 100644 --- a/ld_eventsource/testing/test_error_strategy.py +++ b/ld_eventsource/testing/test_error_strategy.py @@ -37,7 +37,8 @@ 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 @@ -45,14 +46,16 @@ def test_max_time(): should_raise, next2 = next1.apply(err) assert should_raise is False - time.sleep(0.25) + # 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(): - strategy = ErrorStrategy.continue_with_time_limit(0.2) + 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) @@ -60,7 +63,8 @@ def test_max_time_ignores_wall_clock_step(): should_raise, next1 = strategy.apply(err) assert should_raise is False - time.sleep(0.25) + # 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 diff --git a/ld_eventsource/testing/test_sse_client_retry.py b/ld_eventsource/testing/test_sse_client_retry.py index d9f38d8..f94b134 100644 --- a/ld_eventsource/testing/test_sse_client_retry.py +++ b/ld_eventsource/testing/test_sse_client_retry.py @@ -140,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 + 0.05) + # 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)