fix: Clear a stale interrupt flag and time the attempt after the wait

Addresses review feedback on #519.

A new stream connection now clears _interrupted_by_sdk. interrupt() is a
no-op when the connection has already gone, so no Fault arrives to clear
the flag and it could swallow the next genuine server close, losing one
failure and skipping one backoff. Both streaming sources get a test that
fails without the change.

_connection_attempt_start_time is now read after the retry wait instead
of predicted before it, so a clock change during the wait cannot skew the
stream-init latency we report.

Also drops the certificate-classification comment from all four data
sources, says why the bare RetryDelayStrategy is passed to both SSE
clients, and rewords AsyncPollingUpdateProcessor.stop() so it is clear
that the wait after OFF only drains a poll already in flight.

Author: jsonbailey

ldclient/impl/aio/transport.py

@@ -140,7 +140,9 @@ def create(self, url: str, initial_retry_delay: float, query_params=None, sdk_ma
         if proxy:
             aiohttp_request_options["proxy"] = proxy
         if sdk_managed_retry:
-            # The SSE client's retry is disabled; the SDK owns the delay.
+            # The SSE client's retry is disabled; the SDK owns the delay. The base
+            # strategy returns the delay unchanged, so the wait is always zero;
+            # omitting it would select the library's own backoff.
             retry_options: dict = {
                 "initial_retry_delay": 0,
                 "retry_delay_strategy": RetryDelayStrategy(),

ldclient/impl/datasource/async_polling.py

@@ -63,9 +63,10 @@ async def stop(self):
         if self._data_source_update_sink is not None:
             self._data_source_update_sink.update_status(DataSourceState.OFF, None)
 
-        # Wait for the current poll to finish before closing the transport, so we do
-        # not close it while a request is still using it. The close is in a finally
-        # so an owned transport is still released if stop() is cancelled mid-wait.
+        # OFF is reported first, so a listener sees the shutdown at once. The wait
+        # that follows only drains a poll already in flight, so the transport is
+        # not closed while that request still uses it. The close is in a finally,
+        # so an owned transport is released even if stop() is cancelled mid-wait.
         try:
             await self._task.wait_stopped()
         finally:
@@ -95,7 +96,6 @@ async def _fetch_and_store(self) -> None:
             level = log.error if kind is FailureKind.UNEXPECTED else log.warning
             stacktrace = None
         except Exception as e:
-            # A certificate failure lands here too, and is as normal as the rest.
             kind = FailureKind.NORMAL
             error_info = DataSourceErrorInfo(DataSourceErrorKind.UNKNOWN, 0, time.time(), str(e))
             description = "Error encountered when updating flags: %s" % e

ldclient/impl/datasource/async_streaming.py

@@ -89,6 +89,10 @@ async def _run(self):
             self._connection_attempt_start_time = time.time()
             async for action in self._sse.all:
                 if isinstance(action, Start):
+                    # interrupt() is a no-op when the connection has already gone, so
+                    # clear a stale flag here rather than swallow the next real close.
+                    self._interrupted_by_sdk = False
+
                     # On reconnect after an error the timer was cleared; reset it here.
                     # For the initial connect the pre-loop timestamp is already set.
                     if self._connection_attempt_start_time is None:
@@ -256,7 +260,6 @@ async def _handle_error(self, error: Exception) -> bool:
             description = "The server closed the stream connection"
             level = log.warning
         else:
-            # A certificate failure lands here too, and is as normal as the rest.
             kind = FailureKind.NORMAL
             error_info = DataSourceErrorInfo(DataSourceErrorKind.UNKNOWN, 0, time.time(), str(error))
             # no stacktrace here because, for a typical connection error, it'll just be a lengthy tour of HTTP client internals
@@ -270,9 +273,12 @@ async def _handle_error(self, error: Exception) -> bool:
         if self._data_source_update_sink is not None:
             self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, error_info)
 
-        self._connection_attempt_start_time = time.time() + delay
         if delay > 0:
             await asyncio.sleep(delay)
+
+        # Read after the wait, so a clock change during it cannot skew the
+        # stream-init latency we report.
+        self._connection_attempt_start_time = time.time()
         return self._running
 
     # magic methods for "with" statement (used in testing)

ldclient/impl/datasource/polling.py

@@ -100,7 +100,6 @@ def _poll(self) -> None:
             level = log.error if kind is FailureKind.UNEXPECTED else log.warning
             stacktrace = None
         except Exception as e:
-            # A certificate failure lands here too, and is as normal as the rest.
             kind = FailureKind.NORMAL
             error_info = DataSourceErrorInfo(DataSourceErrorKind.UNKNOWN, 0, time.time(), str(e))
             description = "Error encountered when updating flags: %s" % e

ldclient/impl/datasource/streaming.py

@@ -74,6 +74,9 @@ def run(self):
         self._connection_attempt_start_time = time.time()
         for action in self._sse.all:
             if isinstance(action, Start):
+                # interrupt() is a no-op when the connection has already gone, so
+                # clear a stale flag here rather than swallow the next real close.
+                self._interrupted_by_sdk = False
                 record_environment_id(self._data_source_update_sink, action.headers)
             elif isinstance(action, Event):
                 message_ok = False
@@ -137,7 +140,9 @@ def _create_sse_client(self) -> SSEClient:
                 url=self._uri, headers=http_factory.base_headers, pool=stream_http_factory.create_pool_manager(1, self._uri), urllib3_request_options={"timeout": stream_http_factory.timeout}
             ),
             error_strategy=ErrorStrategy.always_continue(),  # we'll make error-handling decisions when we see a Fault
-            # The SSE client's retry is disabled; the SDK owns the delay.
+            # The SSE client's retry is disabled; the SDK owns the delay. The base
+            # strategy returns the delay unchanged, so the wait is always zero;
+            # omitting it would select the library's own backoff.
             initial_retry_delay=0,
             retry_delay_strategy=RetryDelayStrategy(),
             retry_delay_reset_threshold=0,
@@ -230,7 +235,6 @@ def _handle_error(self, error: Exception) -> bool:
             description = "The server closed the stream connection"
             level = log.warning
         else:
-            # A certificate failure lands here too, and is as normal as the rest.
             kind = FailureKind.NORMAL
             error_info = DataSourceErrorInfo(DataSourceErrorKind.UNKNOWN, 0, time.time(), str(error))
             # no stacktrace here because, for a typical connection error, it'll just be a lengthy tour of urllib3 internals
@@ -244,8 +248,12 @@ def _handle_error(self, error: Exception) -> bool:
         if self._data_source_update_sink is not None:
             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(min(delay, TIMEOUT_MAX))
+        interrupted = self._stop_event.wait(min(delay, TIMEOUT_MAX))
+
+        # Read after the wait, so a clock change during it cannot skew the
+        # stream-init latency we report.
+        self._connection_attempt_start_time = time.time()
+        return not interrupted
 
     # magic methods for "with" statement (used in testing)
     def __enter__(self):

ldclient/testing/impl/datasource/test_async_streaming.py

@@ -399,6 +399,30 @@ async def test_our_own_interrupt_is_not_counted_as_a_server_close():
     await proc.stop()
 
 
+@pytest.mark.asyncio
+async def test_a_leaked_interrupt_flag_does_not_swallow_a_server_close():
+    """interrupt() is a no-op when the connection has already gone, so no
+    Fault arrives to clear the flag. A new connection must clear it, or the
+    next genuine close is recorded as ours and the backoff is skipped."""
+    put_data = _make_put_data()
+    actions = [
+        _start(),
+        _event('put', put_data),
+        _fault(error=None),  # a close the SDK did not ask for
+    ]
+
+    retry = _fast_retry_state()
+    proc, _, _, _ = _make_processor(actions, retry_state=retry)
+    proc._interrupted_by_sdk = True
+    proc.start()
+    await _wait_until(lambda: retry._attempts >= 1)
+    await asyncio.sleep(0.1)
+
+    assert retry._attempts == 1
+
+    await proc.stop()
+
+
 @pytest.mark.asyncio
 async def test_unexpected_http_error_keeps_the_processor_running():
     """A rejected SDK key is retried like any other failure. The state never

ldclient/testing/impl/datasource/test_streaming.py

@@ -499,6 +499,34 @@ def listener(s):
                 assert retry._attempts == 1
 
 
+def test_a_leaked_interrupt_flag_does_not_swallow_a_server_close():
+    """interrupt() is a no-op when the connection has already gone, so no
+    Fault arrives to clear the flag. A new connection must clear it, or the
+    next genuine close is recorded as ours and the backoff is skipped."""
+    store = InMemoryFeatureStore()
+    ready = Event()
+    flagv1 = FlagBuilder('flagkey').version(1).build()
+    flagv2 = FlagBuilder('flagkey').version(2).build()
+
+    with start_server() as server:
+        with stream_content(make_put_event([flagv1])) as stream1:
+            with stream_content(make_put_event([flagv2])) as stream2:
+                config = Config(sdk_key='sdk-key', stream_uri=server.uri, initial_reconnect_delay=brief_delay)
+                server.for_path('/all', SequentialHandler(stream1, stream2))
+
+                retry = fast_retry_state()
+                with StreamingUpdateProcessor(config, store, ready, None, retry_state=retry) as sp:
+                    sp._interrupted_by_sdk = True
+                    sp.start()
+                    ready.wait(start_wait)
+                    assert sp.initialized()
+
+                    stream1.close()
+                    expect_update(store, FEATURES, flagv2)
+
+                    assert retry._attempts == 1
+
+
 def _handle_errors_without_waiting(retry, errors):
     """Drives _handle_error for each error and returns nothing. The stop event
     is pre-set so the interruptible wait returns at once."""