fix: Connection close hang on Windows

Author: abeiderman

.github/workflows/ci.yml

@@ -18,6 +18,7 @@ jobs:
   lint:
     name: Lint + Typecheck
     runs-on: ubuntu-latest
+    timeout-minutes: 10
     steps:
       - uses: actions/checkout@v5
 
@@ -33,6 +34,9 @@ jobs:
   test:
     name: Test (Python ${{ matrix.python-version }} on ${{ matrix.os }})
     runs-on: ${{ matrix.os }}
+    # Backstop only. pytest-timeout is what turns a wedged test into a failure with a
+    # thread dump; this bounds anything outside pytest, such as a dependency install.
+    timeout-minutes: 15
     strategy:
       fail-fast: false
       matrix:
@@ -58,6 +62,7 @@ jobs:
   build:
     name: Build distribution
     runs-on: ubuntu-latest
+    timeout-minutes: 10
     steps:
       - uses: actions/checkout@v5
 
@@ -77,6 +82,7 @@ jobs:
   samples:
     name: Sample apps
     runs-on: ubuntu-latest
+    timeout-minutes: 15
     steps:
       - uses: actions/checkout@v5
 

pyproject.toml

@@ -47,6 +47,7 @@ dev = [
   "mypy>=1.14",
   "pytest>=8.3",
   "pytest-cov>=6.0",
+  "pytest-timeout>=2.3",
   "ruff>=0.9",
 ]
 
@@ -99,7 +100,12 @@ enable_error_code = ["redundant-expr", "possibly-undefined", "truthy-bool"]
 
 [tool.pytest.ini_options]
 testpaths = ["tests"]
-addopts = "-ra --strict-markers --strict-config"
+# The timeout covers setup, call and teardown, and is generous next to the longest deliberate
+# wait in the suite (~10s). Several tests park a reader on a real socket, so a threading bug
+# surfaces as a suite that never finishes rather than as a failure; this turns that back into a
+# failure. The thread method is the one that works here: it dumps every thread's stack, and it
+# can end a run that is wedged on a lock held across a blocking read, which SIGALRM cannot.
+addopts = "-ra --strict-markers --strict-config --timeout=60 --timeout-method=thread"
 xfail_strict = true
 
 [tool.coverage.run]

src/configdirector/_eventsource/transport.py

@@ -80,9 +80,30 @@ def close(self) -> None:
         # where no reader is parked on one.
         with contextlib.suppress(ValueError, RuntimeError, OSError):
             self._response.shutdown()
+        # shutdown() finishes the job on POSIX, where it wakes the parked recv(). Winsock only
+        # disallows *subsequent* receives, so on Windows the reader stays in the kernel holding
+        # the response's buffered-reader lock, and the close() below would block acquiring that
+        # same lock -- forever, on whichever thread called close(). closesocket() is what
+        # cancels a pending blocking call, and HTTPConnection.close() drops the socket before it
+        # touches the buffered reader, so it has to run first.
+        _drop_socket(self._response)
         self._response.close()
 
 
+def _drop_socket(response: BaseHTTPResponse) -> None:
+    """Closes the connection's socket, cancelling any read parked on it.
+
+    `_connection` is private, but urllib3 exposes no public way to reach the socket and there is
+    no other way to cancel a blocking read on Windows. A urllib3 that moves or drops it leaves
+    this a no-op, which is the behaviour we had before.
+    """
+    connection = getattr(response, "_connection", None)
+    if connection is None:
+        return
+    with contextlib.suppress(Exception):
+        connection.close()
+
+
 def open_stream(request: StreamRequest) -> ResponseStream:
     response = _pool.request(
         request.method,

tests/eventsource/test_transport.py

@@ -4,15 +4,18 @@
 import threading
 import time
 from collections.abc import Callable, Iterator
+from typing import cast
 
 import pytest
+from urllib3.response import BaseHTTPResponse
 
 from configdirector._eventsource import (
     EventSourceClient,
     EventSourceMessage,
     ReadyState,
     ReconnectionState,
 )
+from configdirector._eventsource.transport import _Stream
 
 from .helpers import wait_for
 
@@ -266,3 +269,69 @@ def handle(request: http.server.BaseHTTPRequestHandler) -> None:
         client.close()
 
     assert received[0].data == payload
+
+
+class TestStreamClose:
+    """The teardown order in `_Stream.close()`, which no live test can pin on a POSIX runner.
+
+    A reader parked in `recv()` holds the response's buffered-reader lock for as long as it is
+    parked. `shutdown()` wakes it on POSIX, so the order is invisible there; Winsock leaves it
+    parked, and closing the response then blocks on that lock forever. Only closing the socket
+    cancels the pending read, so the connection has to go first.
+    """
+
+    class _Connection:
+        def __init__(self, calls: list[str]) -> None:
+            self._calls = calls
+
+        def close(self) -> None:
+            self._calls.append("connection.close")
+
+    class _Response:
+        status = 200
+
+        def __init__(self, calls: list[str], *, shutdown_error: Exception | None = None) -> None:
+            self._calls = calls
+            self._shutdown_error = shutdown_error
+            self._connection = TestStreamClose._Connection(calls)
+
+        def shutdown(self) -> None:
+            self._calls.append("shutdown")
+            if self._shutdown_error is not None:
+                raise self._shutdown_error
+
+        def close(self) -> None:
+            self._calls.append("response.close")
+
+    def test_the_socket_is_dropped_before_the_response_is_closed(self) -> None:
+        calls: list[str] = []
+        _Stream(cast(BaseHTTPResponse, self._Response(calls))).close()
+
+        assert calls == ["shutdown", "connection.close", "response.close"]
+
+    def test_a_refused_shutdown_does_not_skip_the_rest_of_the_teardown(self) -> None:
+        # urllib3 raises when there is no longer a socket to shut down, which is exactly the
+        # case where nothing is parked on one. The response still has to be closed.
+        calls: list[str] = []
+        response = self._Response(calls, shutdown_error=ValueError("no socket"))
+        _Stream(cast(BaseHTTPResponse, response)).close()
+
+        assert calls == ["shutdown", "connection.close", "response.close"]
+
+    def test_a_response_without_a_connection_is_still_closed(self) -> None:
+        # Guards the getattr fallback: a urllib3 that renames `_connection` must degrade to the
+        # old behaviour rather than raising out of close().
+        calls: list[str] = []
+
+        class Bare:
+            status = 200
+
+            def shutdown(self) -> None:
+                calls.append("shutdown")
+
+            def close(self) -> None:
+                calls.append("response.close")
+
+        _Stream(cast(BaseHTTPResponse, Bare())).close()
+
+        assert calls == ["shutdown", "response.close"]