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.

Author: jsonbailey

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()

ld_eventsource/testing/test_http_connect_strategy.py

@@ -256,19 +256,14 @@ 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())
 
     client.close()
 
     created_pool.clear.assert_called_once()
-    connection_pool.close.assert_not_called()

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()
 

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",
 ]