From c74bbb693fc17ad19b51112b804d939c355208be Mon Sep 17 00:00:00 2001
From: jsonbailey <jbailey@launchdarkly.com>
Date: Wed, 15 Jul 2026 16:20:28 -0500
Subject: [PATCH 1/8] fix: Close streaming connection on shutdown instead of
 deferring to GC

Make SSEClient.close()/interrupt() tear down the active streaming
connection deterministically, using a uniform closer gated on urllib3
capabilities:

- urllib3 2.x: resp.shutdown() wakes a reader blocked mid-read, then
  resp.close() sends the TCP FIN and releases the fd.
- urllib3 1.26.x: there is no resp.shutdown() and we do not reach into
  private socket attributes, so we fall back to resp.release_conn(). This
  never hangs, but the socket is not closed deterministically there --
  deterministic close requires urllib3 2.x (the fd is released via GC).

Revert PR #72's iterate-and-close of the pool in _HttpClientImpl.close()
back to a plain PoolManager.clear() for a self-created pool. The explicit
per-connection close() hung on urllib3 1.26.x when a reader was still
blocked mid-read on a connection; this fixes that regression. On 2.x the
active connection is already closed by the connection closer.

Also close the old ConnectionResult on a fault-driven reconnect in
_all_generator (both the normal-end and error paths), matching the async
client, so reconnects tear the previous connection down deterministically
rather than leaving it for garbage collection.

This corrects the pool-ownership/close work from 1.7.1 (SDK-2600) and
targets a 1.7.2 release.
---
 ld_eventsource/http.py                        |  37 +++---
 ld_eventsource/sse_client.py                  |  17 ++-
 .../testing/test_http_connect_strategy.py     |  12 +-
 ...t_http_connect_strategy_with_sse_client.py | 112 ++++++++++++++++++
 .../testing/test_sse_client_retry.py          |  39 ++++++
 5 files changed, 194 insertions(+), 23 deletions(-)

diff --git a/ld_eventsource/http.py b/ld_eventsource/http.py
index 2b94809..c34e4e0 100644
--- a/ld_eventsource/http.py
+++ b/ld_eventsource/http.py
@@ -115,24 +115,31 @@ 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()
+            if hasattr(resp, "shutdown"):
+                # urllib3 2.x: shutdown() wakes a reader blocked mid-read (SHUT_RD) so the
+                # subsequent close() -- which calls self._fp.close() and would otherwise
+                # deadlock on the reader's BufferedReader lock -- can proceed. close() then
+                # sends the TCP FIN and releases the fd. Both are built-ins.
+                try:
+                    resp.shutdown()
+                except Exception:
+                    self.__logger.debug("Error interrupting stream via resp.shutdown()", exc_info=True)
+                resp.close()
+            else:
+                # urllib3 1.26.x has no resp.shutdown(), and we deliberately do not reach
+                # into private socket attributes to find and shut down the socket. Without a
+                # way to wake a blocked reader first, resp.close() could deadlock on the
+                # reader's BufferedReader lock, so we fall back to the original behavior of
+                # releasing the connection back to the pool. This never hangs, but the socket
+                # is not closed deterministically -- deterministic close requires urllib3 2.x.
+                resp.release_conn()
 
         return stream, close, response_headers
 
     def close(self):
         if self.__should_close_pool:
-            # Close pooled connections (sends the TCP FIN) before dropping the pool.
-            # PoolManager.clear() alone drops the pool dict without closing the
-            # underlying sockets, leaving the connection open until garbage collection.
-            for key in list(self.__pool.pools.keys()):
-                connection_pool = self.__pool.pools.get(key)
-                if connection_pool is not None:
-                    try:
-                        connection_pool.close()
-                    except Exception:
-                        self.__logger.debug("Error closing connection pool", exc_info=True)
+            # Only clear a pool we created. On urllib3 2.x the active connection was already
+            # closed by the connection closer (resp.close()); we do not iterate and close the
+            # pool's connections ourselves, because doing so hangs on urllib3 1.26.x when a
+            # reader is still blocked mid-read on a connection.
             self.__pool.clear()
diff --git a/ld_eventsource/sse_client.py b/ld_eventsource/sse_client.py
index 83996cf..76cf0f7 100644
--- a/ld_eventsource/sse_client.py
+++ b/ld_eventsource/sse_client.py
@@ -168,6 +168,19 @@ def interrupt(self):
             self.__connection_result = None
             self._compute_next_retry_delay()
 
+    def _close_current_connection(self):
+        # Close the current connection before dropping it so a fault-driven reconnect tears
+        # the old connection down deterministically (on urllib3 2.x) rather than deferring to
+        # garbage collection. This is only reached after the read loop has ended (stream
+        # exhausted or errored), so the reader is not blocked mid-read and closing cannot
+        # deadlock. Capture into a local and guard for None in case interrupt() nulled it
+        # concurrently; ConnectionResult.close() is idempotent, so a later interrupt()/close()
+        # is safe.
+        result = self.__connection_result
+        self.__connection_result = None
+        if result is not None:
+            result.close()
+
     @property
     def all(self) -> Iterable[Action]:
         """
@@ -218,13 +231,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
 
diff --git a/ld_eventsource/testing/test_http_connect_strategy.py b/ld_eventsource/testing/test_http_connect_strategy.py
index 815dfec..d77f5b6 100644
--- a/ld_eventsource/testing/test_http_connect_strategy.py
+++ b/ld_eventsource/testing/test_http_connect_strategy.py
@@ -255,11 +255,11 @@ 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() clears that pool.
+    # It must NOT iterate and close the individual connection pools itself: on urllib3
+    # 1.26.x that hangs when a reader is still blocked mid-read on a connection. On
+    # urllib3 2.x the active connection is already closed by the connection closer.
     connection_pool = mock.Mock()
     created_pool = mock.MagicMock()
     created_pool.pools.keys.return_value = ['poolkey']
@@ -270,5 +270,5 @@ def test_close_closes_pool_it_created():
 
     client.close()
 
-    connection_pool.close.assert_called_once()
     created_pool.clear.assert_called_once()
+    connection_pool.close.assert_not_called()
diff --git a/ld_eventsource/testing/test_http_connect_strategy_with_sse_client.py b/ld_eventsource/testing/test_http_connect_strategy_with_sse_client.py
index 2502fe7..1364fbe 100644
--- a/ld_eventsource/testing/test_http_connect_strategy_with_sse_client.py
+++ b/ld_eventsource/testing/test_http_connect_strategy_with_sse_client.py
@@ -1,6 +1,11 @@
+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 +63,113 @@ 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'
+
+
+# urllib3 2.x exposes HTTPResponse.shutdown(), which lets the closer wake a reader that is
+# blocked mid-read before closing the connection. urllib3 1.26.x has no such built-in, and
+# under design U-prime we deliberately do not reach into private socket attributes to wake
+# the reader. So on 2.x the reader is woken; on 1.26.x it is not -- but in NEITHER case may
+# the stop call itself hang.
+_CAN_WAKE_READER = 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 on every urllib3 version (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, on any urllib3 version.
+                assert stopped.wait(timeout=5), \
+                    "%s() deadlocked on a concurrently blocked reader" % stop_method_name
+
+                if _CAN_WAKE_READER:
+                    # urllib3 2.x: the closer's resp.shutdown() wakes the blocked reader.
+                    assert reader_woke.wait(timeout=5), \
+                        "%s() did not wake the concurrently blocked reader" % stop_method_name
+                else:
+                    # urllib3 1.26.x: the reader is intentionally not woken (U-prime does not
+                    # touch private socket attrs); only the no-hang guarantee above applies.
+                    pass
+            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
 
diff --git a/ld_eventsource/testing/test_sse_client_retry.py b/ld_eventsource/testing/test_sse_client_retry.py
index fba2fe8..762cee9 100644
--- a/ld_eventsource/testing/test_sse_client_retry.py
+++ b/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(

From 8d0de4ecfc8f5c81dad778ce544d9949f10bc90f Mon Sep 17 00:00:00 2001
From: jsonbailey <jbailey@launchdarkly.com>
Date: Thu, 16 Jul 2026 16:49:42 -0500
Subject: [PATCH 2/8] fix: Require urllib3 2.x and drop the 1.26 stream-close
 fallback

The connection closer now unconditionally calls resp.shutdown() then resp.close() (2.x APIs) to deterministically close the streaming socket; the urllib3 floor is raised to >=2.0.0 and the 1.26 release_conn fallback and its version-gated tests are removed.
---
 ld_eventsource/http.py                        | 29 +++++--------------
 .../testing/test_http_connect_strategy.py     | 11 ++-----
 ...t_http_connect_strategy_with_sse_client.py | 25 ++++------------
 pyproject.toml                                |  4 +--
 4 files changed, 17 insertions(+), 52 deletions(-)

diff --git a/ld_eventsource/http.py b/ld_eventsource/http.py
index c34e4e0..29f47e6 100644
--- a/ld_eventsource/http.py
+++ b/ld_eventsource/http.py
@@ -115,31 +115,16 @@ def connect(self, last_event_id: Optional[str]) -> Tuple[Iterator[bytes], Callab
         stream = resp.stream(_CHUNK_SIZE)
 
         def close():
-            if hasattr(resp, "shutdown"):
-                # urllib3 2.x: shutdown() wakes a reader blocked mid-read (SHUT_RD) so the
-                # subsequent close() -- which calls self._fp.close() and would otherwise
-                # deadlock on the reader's BufferedReader lock -- can proceed. close() then
-                # sends the TCP FIN and releases the fd. Both are built-ins.
-                try:
-                    resp.shutdown()
-                except Exception:
-                    self.__logger.debug("Error interrupting stream via resp.shutdown()", exc_info=True)
-                resp.close()
-            else:
-                # urllib3 1.26.x has no resp.shutdown(), and we deliberately do not reach
-                # into private socket attributes to find and shut down the socket. Without a
-                # way to wake a blocked reader first, resp.close() could deadlock on the
-                # reader's BufferedReader lock, so we fall back to the original behavior of
-                # releasing the connection back to the pool. This never hangs, but the socket
-                # is not closed deterministically -- deterministic close requires urllib3 2.x.
-                resp.release_conn()
+            try:
+                resp.shutdown()
+            except Exception:
+                self.__logger.debug("Error interrupting stream via resp.shutdown()", exc_info=True)
+            resp.close()
 
         return stream, close, response_headers
 
     def close(self):
         if self.__should_close_pool:
-            # Only clear a pool we created. On urllib3 2.x the active connection was already
-            # closed by the connection closer (resp.close()); we do not iterate and close the
-            # pool's connections ourselves, because doing so hangs on urllib3 1.26.x when a
-            # reader is still blocked mid-read on a connection.
+            # Only clear a pool we created; the active connection was already closed by the
+            # connection closer (resp.close()), so clearing the pool is all that's left.
             self.__pool.clear()
diff --git a/ld_eventsource/testing/test_http_connect_strategy.py b/ld_eventsource/testing/test_http_connect_strategy.py
index d77f5b6..0fdb51b 100644
--- a/ld_eventsource/testing/test_http_connect_strategy.py
+++ b/ld_eventsource/testing/test_http_connect_strategy.py
@@ -256,14 +256,10 @@ def test_close_leaves_caller_supplied_pool_open():
 
 
 def test_close_clears_pool_it_created():
-    # When the client creates its own pool (no pool supplied), close() clears that pool.
-    # It must NOT iterate and close the individual connection pools itself: on urllib3
-    # 1.26.x that hangs when a reader is still blocked mid-read on a connection. On
-    # urllib3 2.x the active connection is already closed by the connection closer.
-    connection_pool = mock.Mock()
+    # When the client creates its own pool (no pool supplied), close() clears it. The active
+    # connection is already closed by the connection closer (resp.close()), so clearing the
+    # pool is all that's needed here.
     created_pool = mock.MagicMock()
-    created_pool.pools.keys.return_value = ['poolkey']
-    created_pool.pools.get.return_value = connection_pool
 
     with mock.patch('ld_eventsource.http.PoolManager', return_value=created_pool):
         client = ConnectStrategy.http("http://test").create_client(logger())
@@ -271,4 +267,3 @@ def test_close_clears_pool_it_created():
     client.close()
 
     created_pool.clear.assert_called_once()
-    connection_pool.close.assert_not_called()
diff --git a/ld_eventsource/testing/test_http_connect_strategy_with_sse_client.py b/ld_eventsource/testing/test_http_connect_strategy_with_sse_client.py
index 1364fbe..0a6f0f9 100644
--- a/ld_eventsource/testing/test_http_connect_strategy_with_sse_client.py
+++ b/ld_eventsource/testing/test_http_connect_strategy_with_sse_client.py
@@ -2,8 +2,6 @@
 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 *
@@ -97,18 +95,10 @@ def test_sse_client_reconnects_after_interrupt():
                     assert event2.data == 'data2'
 
 
-# urllib3 2.x exposes HTTPResponse.shutdown(), which lets the closer wake a reader that is
-# blocked mid-read before closing the connection. urllib3 1.26.x has no such built-in, and
-# under design U-prime we deliberately do not reach into private socket attributes to wake
-# the reader. So on 2.x the reader is woken; on 1.26.x it is not -- but in NEITHER case may
-# the stop call itself hang.
-_CAN_WAKE_READER = 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 on every urllib3 version (a hang here is the regression we guard against).
+    # timeout (a hang here is the regression we guard against).
     with start_server() as server:
         with make_stream() as stream:
             server.for_path('/', stream)
@@ -146,18 +136,13 @@ def stopper():
                 stopper_thread = threading.Thread(target=stopper, daemon=True)
                 stopper_thread.start()
 
-                # Must never hang, on any urllib3 version.
+                # Must never hang.
                 assert stopped.wait(timeout=5), \
                     "%s() deadlocked on a concurrently blocked reader" % stop_method_name
 
-                if _CAN_WAKE_READER:
-                    # urllib3 2.x: the closer's resp.shutdown() wakes the blocked reader.
-                    assert reader_woke.wait(timeout=5), \
-                        "%s() did not wake the concurrently blocked reader" % stop_method_name
-                else:
-                    # urllib3 1.26.x: the reader is intentionally not woken (U-prime does not
-                    # touch private socket attrs); only the no-hang guarantee above applies.
-                    pass
+                # The closer's resp.shutdown() wakes the blocked reader.
+                assert reader_woke.wait(timeout=5), \
+                    "%s() did not wake the concurrently blocked reader" % stop_method_name
             finally:
                 client.close()
 
diff --git a/pyproject.toml b/pyproject.toml
index 687c5f0..cb66390 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -22,7 +22,7 @@ classifiers = [
     "Topic :: Software Development :: Libraries",
 ]
 dependencies = [
-    "urllib3>=1.26.0,<3",
+    "urllib3>=2.0.0,<3",
 ]
 
 [project.urls]
@@ -55,7 +55,7 @@ docs = [
     "pyrfc3339>=1.0",
     "jsonpickle>1.4.1",
     "semver>=2.7.9",
-    "urllib3>=1.26.0",
+    "urllib3>=2.0.0",
     "jinja2>=3.1.2",
 ]
 

From 46cb96eede8c012db767ee79e9f1e6a2ef824ff2 Mon Sep 17 00:00:00 2001
From: jsonbailey <jbailey@launchdarkly.com>
Date: Fri, 17 Jul 2026 09:15:06 -0500
Subject: [PATCH 3/8] refactor: Route interrupt() through the shared
 connection-close helper

---
 ld_eventsource/sse_client.py | 11 ++---------
 1 file changed, 2 insertions(+), 9 deletions(-)

diff --git a/ld_eventsource/sse_client.py b/ld_eventsource/sse_client.py
index 76cf0f7..da61d4f 100644
--- a/ld_eventsource/sse_client.py
+++ b/ld_eventsource/sse_client.py
@@ -164,18 +164,11 @@ 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 the current connection before dropping it so a fault-driven reconnect tears
-        # the old connection down deterministically (on urllib3 2.x) rather than deferring to
-        # garbage collection. This is only reached after the read loop has ended (stream
-        # exhausted or errored), so the reader is not blocked mid-read and closing cannot
-        # deadlock. Capture into a local and guard for None in case interrupt() nulled it
-        # concurrently; ConnectionResult.close() is idempotent, so a later interrupt()/close()
-        # is safe.
+        # Close and drop the current connection, if any.
         result = self.__connection_result
         self.__connection_result = None
         if result is not None:

From cd3a346b65af20b3ddf769945162cb1a171582c0 Mon Sep 17 00:00:00 2001
From: jsonbailey <jbailey@launchdarkly.com>
Date: Tue, 21 Jul 2026 17:35:41 -0500
Subject: [PATCH 4/8] fix: Close connection socket directly to wake blocked
 reader on Windows

resp.shutdown() (SHUT_RD) wakes a blocked reader on POSIX but not Windows. Closing the connection's socket via the public resp.connection.close() wakes it on Windows too and bypasses the BufferedReader lock that resp.close() deadlocks on, using only public API and without touching the caller's pool.
---
 ld_eventsource/http.py | 8 ++++++++
 1 file changed, 8 insertions(+)

diff --git a/ld_eventsource/http.py b/ld_eventsource/http.py
index 29f47e6..152a830 100644
--- a/ld_eventsource/http.py
+++ b/ld_eventsource/http.py
@@ -119,6 +119,14 @@ def close():
                 resp.shutdown()
             except Exception:
                 self.__logger.debug("Error interrupting stream via resp.shutdown()", exc_info=True)
+            # Close the connection's socket directly (public API) to also wake a
+            # blocked reader on Windows, where resp.shutdown() does not interrupt recv().
+            conn = resp.connection
+            if conn is not None:
+                try:
+                    conn.close()
+                except Exception:
+                    self.__logger.debug("Error closing stream connection", exc_info=True)
             resp.close()
 
         return stream, close, response_headers

From 75c822a4f193ff5c79258064988bdd4867257a33 Mon Sep 17 00:00:00 2001
From: jsonbailey <jbailey@launchdarkly.com>
Date: Wed, 22 Jul 2026 11:18:40 -0500
Subject: [PATCH 5/8] fix: Gate deterministic stream close by platform and
 urllib3 capability

Deterministic close (shutdown+close) only on POSIX with urllib3>=2.3 (which has HTTPResponse.shutdown); Windows and older urllib3 fall back to release_conn (leak, never hang). Keeps urllib3>=1.26 support, restores owned-pool connection close, and gates the reader-woke test assertion accordingly.
---
 ld_eventsource/http.py                        | 40 ++++++++++++-------
 .../testing/test_http_connect_strategy.py     | 10 +++--
 ...t_http_connect_strategy_with_sse_client.py | 16 ++++++--
 pyproject.toml                                |  4 +-
 4 files changed, 47 insertions(+), 23 deletions(-)

diff --git a/ld_eventsource/http.py b/ld_eventsource/http.py
index 152a830..4a369fb 100644
--- a/ld_eventsource/http.py
+++ b/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,24 +116,33 @@ def connect(self, last_event_id: Optional[str]) -> Tuple[Iterator[bytes], Callab
         stream = resp.stream(_CHUNK_SIZE)
 
         def close():
-            try:
-                resp.shutdown()
-            except Exception:
-                self.__logger.debug("Error interrupting stream via resp.shutdown()", exc_info=True)
-            # Close the connection's socket directly (public API) to also wake a
-            # blocked reader on Windows, where resp.shutdown() does not interrupt recv().
-            conn = resp.connection
-            if conn is not None:
-                try:
-                    conn.close()
-                except Exception:
-                    self.__logger.debug("Error closing stream connection", exc_info=True)
-            resp.close()
+            # 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
 
     def close(self):
         if self.__should_close_pool:
-            # Only clear a pool we created; the active connection was already closed by the
-            # connection closer (resp.close()), so clearing the pool is all that's left.
+            # Close pooled connections (sends the TCP FIN) before dropping the pool.
+            # PoolManager.clear() alone drops the pool dict without closing the
+            # underlying sockets, leaving the connection open until garbage collection.
+            for key in list(self.__pool.pools.keys()):
+                connection_pool = self.__pool.pools.get(key)
+                if connection_pool is not None:
+                    try:
+                        connection_pool.close()
+                    except Exception:
+                        self.__logger.debug("Error closing connection pool", exc_info=True)
             self.__pool.clear()
diff --git a/ld_eventsource/testing/test_http_connect_strategy.py b/ld_eventsource/testing/test_http_connect_strategy.py
index 0fdb51b..e0305c8 100644
--- a/ld_eventsource/testing/test_http_connect_strategy.py
+++ b/ld_eventsource/testing/test_http_connect_strategy.py
@@ -256,14 +256,18 @@ def test_close_leaves_caller_supplied_pool_open():
 
 
 def test_close_clears_pool_it_created():
-    # When the client creates its own pool (no pool supplied), close() clears it. The active
-    # connection is already closed by the connection closer (resp.close()), so clearing the
-    # pool is all that's needed here.
+    # 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']
+    created_pool.pools.get.return_value = connection_pool
 
     with mock.patch('ld_eventsource.http.PoolManager', return_value=created_pool):
         client = ConnectStrategy.http("http://test").create_client(logger())
 
     client.close()
 
+    connection_pool.close.assert_called_once()
     created_pool.clear.assert_called_once()
diff --git a/ld_eventsource/testing/test_http_connect_strategy_with_sse_client.py b/ld_eventsource/testing/test_http_connect_strategy_with_sse_client.py
index 0a6f0f9..69f0b42 100644
--- a/ld_eventsource/testing/test_http_connect_strategy_with_sse_client.py
+++ b/ld_eventsource/testing/test_http_connect_strategy_with_sse_client.py
@@ -1,7 +1,10 @@
+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 *
@@ -95,6 +98,12 @@ def test_sse_client_reconnects_after_interrupt():
                     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
@@ -140,9 +149,10 @@ def stopper():
                 assert stopped.wait(timeout=5), \
                     "%s() deadlocked on a concurrently blocked reader" % stop_method_name
 
-                # The closer's resp.shutdown() wakes the blocked reader.
-                assert reader_woke.wait(timeout=5), \
-                    "%s() did not wake the 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()
 
diff --git a/pyproject.toml b/pyproject.toml
index cb66390..687c5f0 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -22,7 +22,7 @@ classifiers = [
     "Topic :: Software Development :: Libraries",
 ]
 dependencies = [
-    "urllib3>=2.0.0,<3",
+    "urllib3>=1.26.0,<3",
 ]
 
 [project.urls]
@@ -55,7 +55,7 @@ docs = [
     "pyrfc3339>=1.0",
     "jsonpickle>1.4.1",
     "semver>=2.7.9",
-    "urllib3>=2.0.0",
+    "urllib3>=1.26.0",
     "jinja2>=3.1.2",
 ]
 

From 12836c89d20b78081706477674ec55073f163e63 Mon Sep 17 00:00:00 2001
From: jsonbailey <jbailey@launchdarkly.com>
Date: Wed, 22 Jul 2026 11:32:24 -0500
Subject: [PATCH 6/8] ci: Add lowest-direct dependency floor job (linux +
 windows)

Regular jobs run against locked/latest deps; this guards the minimum declared versions (notably urllib3 1.26, which lacks HTTPResponse.shutdown()) on both platforms. One Python, one lowest-direct resolution -- not a full version matrix.
---
 .github/workflows/ci.yml | 22 ++++++++++++++++++++++
 1 file changed, 22 insertions(+)

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index d47af75..f2ef220 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -77,3 +77,25 @@ jobs:
 
       - name: Run tests
         run: make test
+
+  # Guards the minimum declared dependency versions (e.g. urllib3 1.26, which
+  # lacks HTTPResponse.shutdown()) on both platforms, since the regular jobs run
+  # against the locked/latest versions. One Python; not a full version matrix.
+  floor:
+    name: floor (lowest-direct deps)
+    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 against lowest-direct dependency versions
+        run: uv run --resolution lowest-direct --all-extras pytest ld_eventsource/testing

From 8ae10d12d917a86c51d13bcc3496703dd6ccb994 Mon Sep 17 00:00:00 2001
From: jsonbailey <jbailey@launchdarkly.com>
Date: Wed, 22 Jul 2026 11:53:06 -0500
Subject: [PATCH 7/8] ci: Use make test with UV_RESOLUTION in the floor job

Reuses the canonical test target (PYTEST_FLAGS incl. -W error::SyntaxWarning) instead of a hardcoded pytest invocation; UV_RESOLUTION=lowest-direct applies to both the uv sync and uv run pytest steps.
---
 .github/workflows/ci.yml | 8 ++++++--
 1 file changed, 6 insertions(+), 2 deletions(-)

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index f2ef220..f5ee93d 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -83,6 +83,10 @@ jobs:
   # against the locked/latest versions. One Python; not a full version matrix.
   floor:
     name: floor (lowest-direct deps)
+    # uv reads UV_RESOLUTION, so both `uv sync` (install) and `uv run pytest`
+    # in `make test` resolve the minimum declared versions.
+    env:
+      UV_RESOLUTION: lowest-direct
     strategy:
       fail-fast: false
       matrix:
@@ -97,5 +101,5 @@ jobs:
         with:
           python-version: "3.10"
 
-      - name: Run tests against lowest-direct dependency versions
-        run: uv run --resolution lowest-direct --all-extras pytest ld_eventsource/testing
+      - name: Run tests
+        run: make test

From 6643d7972276a9408bd1a6cb54349b4bce09e96f Mon Sep 17 00:00:00 2001
From: jsonbailey <jbailey@launchdarkly.com>
Date: Wed, 22 Jul 2026 11:59:43 -0500
Subject: [PATCH 8/8] ci: Simplify floor job comments

---
 .github/workflows/ci.yml | 7 ++-----
 1 file changed, 2 insertions(+), 5 deletions(-)

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index f5ee93d..c6d988e 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -78,13 +78,10 @@ jobs:
       - name: Run tests
         run: make test
 
-  # Guards the minimum declared dependency versions (e.g. urllib3 1.26, which
-  # lacks HTTPResponse.shutdown()) on both platforms, since the regular jobs run
-  # against the locked/latest versions. One Python; not a full version matrix.
+  # Guards the minimum declared dependency versions, since the regular jobs
+  # run against the locked/latest versions. 
   floor:
     name: floor (lowest-direct deps)
-    # uv reads UV_RESOLUTION, so both `uv sync` (install) and `uv run pytest`
-    # in `make test` resolve the minimum declared versions.
     env:
       UV_RESOLUTION: lowest-direct
     strategy: