fix: Close streaming connections deterministically where the platform allows

.github/workflows/ci.yml

@@ -77,3 +77,26 @@ jobs:
 
       - name: Run tests
         run: make test
+
+  # Guards the minimum declared dependency versions, since the regular jobs
+  # run against the locked/latest versions. 
+  floor:
+    name: floor (lowest-direct deps)
+    env:
+      UV_RESOLUTION: lowest-direct
+    strategy:
+      fail-fast: false
+      matrix:
+        os: [ubuntu-latest, windows-latest]
+    runs-on: ${{ matrix.os }}
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Set up uv
+        uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # v8.2.0
+        with:
+          python-version: "3.10"
+
+      - name: Run tests
+        run: make test

ld_eventsource/http.py

@@ -1,3 +1,4 @@
+import sys
 from logging import Logger
 from typing import Any, Callable, Dict, Iterator, Optional, Tuple, cast
 from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
@@ -115,11 +116,20 @@ def connect(self, last_event_id: Optional[str]) -> Tuple[Iterator[bytes], Callab
         stream = resp.stream(_CHUNK_SIZE)
 
         def close():
-            try:
-                resp.shutdown()
-            except Exception:
-                pass
-            resp.release_conn()
+            # We can only deterministically close the socket where a reader blocked
+            # mid-read can first be woken: POSIX with urllib3's resp.shutdown()
+            # (SHUT_RD). On Windows (shutdown() does not interrupt recv()) or on
+            # urllib3 without shutdown(), release instead -- the socket leaks until
+            # GC but this never deadlocks on the reader's BufferedReader lock.
+            if sys.platform != "win32" and hasattr(resp, "shutdown"):
+                if resp.connection is not None:
+                    try:
+                        resp.shutdown()
+                    except Exception:
+                        self.__logger.debug("Error interrupting stream via resp.shutdown()", exc_info=True)
+                resp.close()
+            else:
+                resp.release_conn()
 
         return stream, close, response_headers
 

ld_eventsource/sse_client.py

@@ -164,10 +164,16 @@ def interrupt(self):
         """
         if self.__connection_result:
             self.__interrupted = True
-            self.__connection_result.close()
-            self.__connection_result = None
+            self._close_current_connection()
             self._compute_next_retry_delay()
 
+    def _close_current_connection(self):
+        # Close and drop the current connection, if any.
+        result = self.__connection_result
+        self.__connection_result = None
+        if result is not None:
+            result.close()
+
     @property
     def all(self) -> Iterable[Action]:
         """
@@ -218,13 +224,13 @@ def _all_generator(self):
                         break
                 # If we finished iterating all of reader.events_and_comments(), it means the stream
                 # was closed without an error.
-                self.__connection_result = None
+                self._close_current_connection()
             except Exception as e:
                 if self.__closed:
                     # It's normal to get an I/O error if we force-closed the stream to shut down
                     return
                 error = e
-                self.__connection_result = None
+                self._close_current_connection()
             finally:
                 self.__last_event_id = reader.last_event_id
 

ld_eventsource/testing/test_http_connect_strategy.py

@@ -255,11 +255,10 @@ def test_close_leaves_caller_supplied_pool_open():
     pool.clear.assert_not_called()
 
 
-def test_close_closes_pool_it_created():
-    # When the client creates its own pool (no pool supplied), close() must close
-    # the pooled connections synchronously -- sending the TCP FIN now -- rather
-    # than leaving the sockets open until garbage collection. PoolManager.clear()
-    # alone does not close them on urllib3 2.x.
+def test_close_clears_pool_it_created():
+    # When the client creates its own pool (no pool supplied), close() closes the pooled
+    # connections (to send FIN, since PoolManager.clear() does not on urllib3 2.x) and
+    # then clears the pool.
     connection_pool = mock.Mock()
     created_pool = mock.MagicMock()
     created_pool.pools.keys.return_value = ['poolkey']

ld_eventsource/testing/test_http_connect_strategy_with_sse_client.py

@@ -1,6 +1,12 @@
+import sys
+import threading
+import time
 from urllib.parse import parse_qsl
 
+import urllib3.response
+
 from ld_eventsource import *
+from ld_eventsource.actions import *
 from ld_eventsource.config import *
 from ld_eventsource.testing.helpers import *
 from ld_eventsource.testing.http_util import *
@@ -58,6 +64,107 @@ def test_sse_client_reconnects_after_socket_closed():
                     assert event2.data == 'data2'
 
 
+def test_sse_client_reconnects_after_interrupt():
+    # interrupt() now fully closes the active HTTP connection (resp.close()) rather than
+    # releasing it back to the pool, so this verifies the client can still open a fresh
+    # stream and reconnect after an interrupt.
+    with start_server() as server:
+        with make_stream() as stream1:
+            with make_stream() as stream2:
+                server.for_path('/', SequentialHandler(stream1, stream2))
+                stream1.push("event: a\ndata: data1\n\n")
+                stream2.push("event: b\ndata: data2\n\n")
+                with SSEClient(
+                    connect=ConnectStrategy.http(server.uri),
+                    error_strategy=ErrorStrategy.always_continue(),
+                    initial_retry_delay=0,
+                ) as client:
+                    client.start()
+                    event1 = next(client.events)
+                    assert event1.event == 'a'
+                    assert event1.data == 'data1'
+
+                    # interrupt() closes the active connection (resp.close()); the
+                    # client discards it and will open a fresh stream on the next read.
+                    client.interrupt()
+
+                    # Release the first server-side handler so this single-threaded test
+                    # server can accept the reconnect. (The reconnect itself is driven by
+                    # the interrupt above, not by this close.)
+                    stream1.close()
+
+                    event2 = next(client.events)
+                    assert event2.event == 'b'
+                    assert event2.data == 'data2'
+
+
+# resp.shutdown() (SHUT_RD) can wake a reader blocked mid-recv only on POSIX with
+# urllib3 >= 2.3 (which has HTTPResponse.shutdown()). Elsewhere the closer releases the
+# connection instead, so the reader is NOT woken -- but the stop must still never hang.
+_CAN_WAKE_READER = sys.platform != "win32" and hasattr(urllib3.response.HTTPResponse, "shutdown")
+
+
+def _run_concurrent_stop_test(stop_method_name):
+    # Exercises the concurrent shutdown pattern: a worker thread blocks mid-read on an idle
+    # stream while another thread stops the client. The stop call must return within the
+    # timeout (a hang here is the regression we guard against).
+    with start_server() as server:
+        with make_stream() as stream:
+            server.for_path('/', stream)
+            stream.push("event: a\ndata: data1\n\n")
+            client = SSEClient(
+                connect=ConnectStrategy.http(server.uri),
+                error_strategy=ErrorStrategy.always_continue(),
+                initial_retry_delay=0,
+            )
+            try:
+                client.start()
+                event1 = next(client.events)
+                assert event1.data == 'data1'
+
+                reader_woke = threading.Event()
+
+                def reader():
+                    try:
+                        next(client.all)  # blocks mid-read on the idle stream
+                    except BaseException:
+                        pass
+                    finally:
+                        reader_woke.set()
+
+                reader_thread = threading.Thread(target=reader, daemon=True)
+                reader_thread.start()
+                time.sleep(0.3)  # let the reader block inside the socket read
+
+                stopped = threading.Event()
+
+                def stopper():
+                    getattr(client, stop_method_name)()
+                    stopped.set()
+
+                stopper_thread = threading.Thread(target=stopper, daemon=True)
+                stopper_thread.start()
+
+                # Must never hang.
+                assert stopped.wait(timeout=5), \
+                    "%s() deadlocked on a concurrently blocked reader" % stop_method_name
+
+                if _CAN_WAKE_READER:
+                    # POSIX + urllib3>=2.3: the closer's resp.shutdown() wakes the reader.
+                    assert reader_woke.wait(timeout=5), \
+                        "%s() did not wake the concurrently blocked reader" % stop_method_name
+            finally:
+                client.close()
+
+
+def test_close_stops_without_hang_with_concurrent_reader():
+    _run_concurrent_stop_test('close')
+
+
+def test_interrupt_stops_without_hang_with_concurrent_reader():
+    _run_concurrent_stop_test('interrupt')
+
+
 def test_sse_client_allows_modifying_query_params_dynamically():
     count = 0
 

ld_eventsource/testing/test_sse_client_retry.py

@@ -154,6 +154,45 @@ def test_retry_delay_gets_reset_after_threshold():
         assert client.next_retry_delay == initial_delay * 2
 
 
+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.
+    closed = []
+
+    class RespondAndTrackClose(MockConnectionHandler):
+        def __init__(self, data, index):
+            self.__data = data
+            self.__index = index
+
+        def apply(self) -> ConnectionResult:
+            index = self.__index
+            return ConnectionResult(
+                stream=iter([bytes(self.__data, 'utf-8')]),
+                closer=lambda: closed.append(index),
+            )
+
+    mock = MockConnectStrategy(
+        RespondAndTrackClose("data: data1\n\n", 0),
+        RespondAndTrackClose("data: data2\n\n", 1),
+        ExpectNoMoreRequests(),
+    )
+    with SSEClient(
+        connect=mock,
+        error_strategy=ErrorStrategy.always_continue(),
+        retry_delay_strategy=no_delay(),
+    ) as client:
+        events = client.events
+
+        event1 = next(events)
+        assert event1.data == 'data1'
+
+        # Reading data2 requires the first stream to end and the client to reconnect;
+        # the first connection must have been closed as part of that reconnect.
+        event2 = next(events)
+        assert event2.data == 'data2'
+        assert closed == [0]
+
+
 def test_can_interrupt_and_restart_stream():
     initial_delay = 0.005
     mock = MockConnectStrategy(