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..4c21a3b 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 @@ -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 diff --git a/ld_eventsource/testing/test_error_strategy.py b/ld_eventsource/testing/test_error_strategy.py index ed21b2a..336c96c 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 * @@ -35,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 @@ -43,7 +46,25 @@ def test_max_time(): 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 diff --git a/ld_eventsource/testing/test_sse_client_retry.py b/ld_eventsource/testing/test_sse_client_retry.py index 762cee9..f94b134 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 * @@ -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.