Stop reading when the opening handshake fails.

docs/project/changelog.rst

@@ -48,6 +48,9 @@ Bug fixes
 * Prevented ``Frame.__str__`` from crashing when a text frame is fragmented in
   the middle of a UTF-8 sequence.
 
+* Fixed a race condition that could delay closing connections in the
+  :mod:`threading` implementation.
+
 .. _16.0:
 
 16.0

src/websockets/sync/connection.py

@@ -20,7 +20,7 @@
 )
 from ..frames import DATA_OPCODES, CloseCode, Frame, Opcode
 from ..http11 import Request, Response
-from ..protocol import CLOSED, OPEN, Event, Protocol, State
+from ..protocol import CLOSED, CONNECTING, OPEN, Event, Protocol, State
 from ..typing import BytesLike, Data, DataLike, LoggerLike, Subprotocol
 from .messages import Assembler
 from .utils import Deadline
@@ -867,6 +867,14 @@ def recv_events(self) -> None:
                         if self.close_deadline is None:
                             self.close_deadline = Deadline(self.close_timeout)
 
+                    # If the opening handshake failed, the connection never
+                    # reached the OPEN state and there's no closing handshake
+                    # to wait for. Remember this so we can stop reading below.
+                    handshake_failed = (
+                        self.protocol.state is CONNECTING
+                        and self.protocol.close_expected()
+                    )
+
                 # Unlock conn_mutex before processing events. Else, the
                 # application can't send messages in response to events.
 
@@ -878,6 +886,17 @@ def recv_events(self) -> None:
                     # This isn't expected to raise an exception.
                     self.process_event(event)
 
+                # When the opening handshake fails, the connection is unusable
+                # and close_socket() runs as soon as the handshake failure
+                # propagates. Stop reading instead of waiting for the peer to
+                # close the TCP connection, because closing a socket in one
+                # thread doesn't always interrupt recv() in another thread --
+                # e.g. on macOS, when the protocol already half-closed the
+                # socket, shutdown() fails with ENOTCONN and close() doesn't
+                # interrupt recv().
+                if handshake_failed:
+                    break
+
             # Breaking out of the while True: ... loop means that we believe
             # that the socket doesn't work anymore.
 

tests/sync/test_client.py

@@ -1,3 +1,4 @@
+import errno
 import http
 import logging
 import os
@@ -190,6 +191,79 @@ def close_connection(self, request):
                 "connection closed while reading HTTP status line",
             )
 
+    def test_connection_closed_promptly_after_invalid_status_response(self):
+        """Client closes the connection promptly when the handshake fails."""
+
+        class HalfClosingSocket(socket.socket):
+            """
+            Reproduces the shutdown()/close() semantics observed on macOS:
+            once the protocol half-closes the socket for writing, a further
+            shutdown() raises ENOTCONN instead of succeeding, and close()
+            doesn't interrupt a recv() call already blocked in another
+            thread.
+
+            """
+
+            def __init__(self, plain):
+                super().__init__(fileno=plain.detach())
+                self.half_closed = False
+
+            def shutdown(self, how):
+                if how == socket.SHUT_WR and not self.half_closed:
+                    self.half_closed = True
+                    super().shutdown(how)
+                else:
+                    raise OSError(errno.ENOTCONN, "Socket is not connected")
+
+            def close(self):
+                pass  # doesn't interrupt a recv() blocked in another thread
+
+            def really_close(self):
+                super().close()
+
+        response = (
+            b"HTTP/1.1 302 Found\r\n"
+            b"Location: http://localhost/\r\n"
+            b"Content-Length: 0\r\n"
+            b"\r\n"
+        )
+        keep_open = threading.Event()
+
+        def keep_connection_open():
+            conn, _ = server.accept()
+            with conn:
+                request = b""
+                while b"\r\n\r\n" not in request:
+                    data = conn.recv(4096)
+                    if data == b"":  # pragma: no cover
+                        return
+                    request += data
+                conn.sendall(response)
+                # Keep the TCP connection open past the handshake failure.
+                keep_open.wait()
+
+        with socket.create_server(("localhost", 0)) as server:
+            host, port = server.getsockname()
+
+            thread = threading.Thread(target=keep_connection_open)
+            thread.start()
+            self.addCleanup(thread.join)
+            self.addCleanup(keep_open.set)
+
+            plain = socket.create_connection((host, port))
+            sock = HalfClosingSocket(plain)
+            self.addCleanup(sock.really_close)
+
+            t0 = time.time()
+            with self.assertRaises(InvalidStatus):
+                with connect(f"ws://{host}:{port}/", sock=sock, close_timeout=20 * MS):
+                    self.fail("did not raise")
+            t1 = time.time()
+
+        # The client doesn't wait anywhere close to close_timeout because it
+        # doesn't wait for the peer to close the connection.
+        self.assertLess(t1 - t0, 10 * MS)
+
     def test_http_response(self):
         """Client reads HTTP response."""
 

tests/sync/test_connection.py

@@ -655,6 +655,33 @@ def fragments():
         )
         self.assertIsNone(exc.__cause__)
 
+    # Test closing the connection when the opening handshake fails.
+
+    def test_stops_reading_when_the_opening_handshake_fails(self):
+        """recv_events stops reading when the opening handshake fails."""
+        socket_, remote_socket = socket.socketpair()
+        self.addCleanup(remote_socket.close)
+        # Simulate the state of the protocol after the opening handshake
+        # failed: EOF was sent, incoming data is discarded, and the state
+        # is still CONNECTING. See ClientProtocol.parse and
+        # ServerProtocol.parse.
+        protocol = Protocol(self.LOCAL, state=State.CONNECTING)
+        protocol.send_eof()
+        protocol.parser = protocol.discard()
+        next(protocol.parser)  # start coroutine
+        connection = Connection(socket_, protocol, close_timeout=20 * MS)
+        self.addCleanup(connection.close_socket)
+
+        t0 = time.time()
+        remote_socket.sendall(b"remaining data to discard")
+        connection.recv_events_thread.join(1)
+        t1 = time.time()
+
+        # The connection closes promptly instead of waiting for the peer to
+        # close the TCP connection.
+        self.assertFalse(connection.recv_events_thread.is_alive())
+        self.assertLess(t1 - t0, 10 * MS)
+
     # Test ping.
 
     @patch("random.getrandbits")