Terminate quickly when opening handshake fails on macOS.

docs/project/changelog.rst

@@ -117,6 +117,9 @@ Bug fixes
 
 * Restored compatibility of the ``websockets`` CLI with Windows.
 
+* Fixed a bug that could delay or block the client in the :mod:`threading`
+  implementation on macOS when the opening handshake fails.
+
 .. _16.1.1:
 
 16.1.1

src/websockets/sync/connection.py

@@ -22,7 +22,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
@@ -815,8 +815,19 @@ def recv_events(self) -> None:
         ``recv_events()`` exits immediately when ``self.socket`` is closed.
 
         """
+        # When the opening handshake fails, we cannot trust rules in RFC 6455
+        # for closing TCP connections will be followed. The HTTP server could
+        # keep the connection alive after sending the response. We attempt to
+        # close the connection immediately. Unfortunately, this is unreliable
+        # on macOS; recv() may block until close_timeout elapses:
+        # https://github.com/python-websockets/websockets/issues/1596
+        # https://github.com/python/cpython/issues/154224
+        # In that case, break out of the loop to prevent recv() from blocking
+        # until close_timeout elapses.
+        close_expected_while_connecting = False
+
         try:
-            while True:
+            while not close_expected_while_connecting:
                 try:
                     # If the assembler buffer is full, block until it drains.
                     with self.recv_flow_control:
@@ -869,6 +880,10 @@ def recv_events(self) -> None:
                     if self.protocol.close_expected():
                         if self.close_deadline is None:
                             self.close_deadline = Deadline(self.close_timeout)
+                        # Ignore coverage because the opening handshake is
+                        # tested in test_client.py, not test_connection.py.
+                        if self.protocol.state is CONNECTING:  # pragma: no cover
+                            close_expected_while_connecting = True
 
                 # Unlock conn_mutex before processing events. Else, the
                 # application can't send messages in response to events.
@@ -881,8 +896,8 @@ def recv_events(self) -> None:
                     # This isn't expected to raise an exception.
                     self.process_event(event)
 
-            # Breaking out of the while True: ... loop means that we believe
-            # that the socket doesn't work anymore.
+            # Breaking out of the while not close_expected_while_connecting: ...
+            # loop means that we believe that the socket doesn't work anymore.
 
             with self.protocol_mutex:
                 # Feed the end of the data stream to the protocol.

tests/asyncio/test_client.py

@@ -488,7 +488,7 @@ def http_response(connection, request):
         self.assertEqual(raised.exception.response.status_code, 200)
         self.assertEqual(raised.exception.response.body.decode(), "👌")
 
-    async def test_junk_handshake(self):
+    async def test_junk_handshake_response(self):
         """Client closes the connection when receiving non-HTTP response from server."""
 
         async def junk(reader, writer):

tests/asyncio/test_server.py

@@ -467,7 +467,7 @@ async def test_connection_closed_during_handshake(self):
             "connection closed while reading HTTP request line",
         )
 
-    async def test_junk_handshake(self):
+    async def test_junk_handshake_request(self):
         """Server closes the connection when receiving non-HTTP request from client."""
         with self.assertLogs("websockets", logging.DEBUG) as logs:
             async with serve(*args) as server:

tests/sync/test_client.py

@@ -220,7 +220,45 @@ def http_response(connection, request):
         self.assertEqual(raised.exception.response.status_code, 200)
         self.assertEqual(raised.exception.response.body.decode(), "👌")
 
-    def test_junk_handshake(self):
+    def test_http_response_with_keep_alive(self):
+        """Client reads HTTP response and server keeps connection alive."""
+
+        class KeepAliveHandler(socketserver.BaseRequestHandler):
+            def handle(self):
+                time.sleep(MS)  # wait for the client to send the handshake request
+                self.request.recv(4096)  # assume we read the full handshake request
+                self.request.send(
+                    b"HTTP/1.1 410 Gone\r\n"
+                    b"Connection: keep-alive\r\n"  # default in HTTP/1.1
+                    b"Content-Length: 0\r\n"  # avoids "read body until EOF"
+                    b"\r\n"
+                )
+                # Wait for the client to close the connection.
+                self.request.recv(4096)
+                self.request.close()
+
+        server = socketserver.TCPServer(("localhost", 0), KeepAliveHandler)
+        host, port = server.server_address
+        with server:
+            thread = threading.Thread(target=server.serve_forever, args=(MS,))
+            thread.start()
+            try:
+                t0 = time.time()
+                with self.assertRaises(InvalidStatus) as raised:
+                    with connect(f"ws://{host}:{port}"):
+                        self.fail("did not raise")
+                t1 = time.time()
+            finally:
+                server.shutdown()
+                thread.join()
+
+        self.assertEqual(
+            str(raised.exception),
+            "server rejected WebSocket connection: HTTP 410",
+        )
+        self.assertLess(t1 - t0, 5 * MS)
+
+    def test_junk_handshake_response(self):
         """Client closes the connection when receiving non-HTTP response from server."""
 
         class JunkHandler(socketserver.BaseRequestHandler):

tests/sync/test_connection.py

@@ -1000,6 +1000,30 @@ def test_unexpected_failure_in_send_context(self, send_text):
             self.connection.send("😀")
         self.assertIsInstance(raised.exception.__cause__, AssertionError)
 
+    # Test recv_events().
+
+    def test_close_expected_during_connecting(self):
+        """recv_events() terminates without waiting for close_timeout."""
+        # See https://github.com/python-websockets/websockets/issues/1596.
+        socket_, remote_socket = socket.socketpair()
+        protocol = Protocol(self.LOCAL, state=State.CONNECTING)
+        # Simulate ClientProtocol / ServerProtocol deciding that the TCP
+        # connection must close because the opening handshake failed.
+        protocol.send_eof()
+        connection = Connection(socket_, protocol, close_timeout=5 * MS)
+        try:
+            t0 = time.time()
+            remote_socket.sendall(b"\x00")
+            connection.recv_events_thread.join(3 * MS)
+            t1 = time.time()
+
+            self.assertFalse(connection.recv_events_thread.is_alive())
+            self.assertLess(t1 - t0, 3 * MS)
+        finally:
+            remote_socket.close()
+            connection.close_socket()
+            connection.recv_events_thread.join()
+
     # Test broadcast.
 
     def test_broadcast_text(self):

tests/sync/test_server.py

@@ -349,7 +349,7 @@ def test_connection_closed_during_handshake(self):
             "connection closed while reading HTTP request line",
         )
 
-    def test_junk_handshake(self):
+    def test_junk_handshake_request(self):
         """Server closes the connection when receiving non-HTTP request from client."""
         with self.assertLogs("websockets.server", logging.DEBUG) as logs:
             with run_server() as server: