WSS: Fix heartbeat timeout logic

Make sure to unblock the `receive` operation by feeding the receiver an error in a `WSMessage`

Change `TimeoutError` to `ServerTimeoutError` to accurately represent failure (this is backwards compatible since `ServerTimeoutError` has `TimeoutError` in the MRO)

fixes #8540

CHANGES/8540.bugfix.rst

@@ -0,0 +1,7 @@
+WSS: Fix heartbeat timeout logic -- by :user:`arcivanov`.
+
+The :py:meth:`~aiohttp.client_ws.ClientWebSocketResponse._pong_not_received` did not
+ terminate a :py:meth:`~aiohttp.client_ws.ClientWebSocketResponse.receive` operation.
+ This change causes `_pong_not_received` to feed the `reader` an error message, causing
+ pending `receive` to terminate and return the error message. The error message contains
+ the exception :py:class:`~aiohttp.client_exceptions.ServerTimeoutError`.

aiohttp/client_ws.py

@@ -5,7 +5,7 @@
 import sys
 from typing import Any, Final, Optional, cast
 
-from .client_exceptions import ClientError
+from .client_exceptions import ClientError, ServerTimeoutError
 from .client_reqrep import ClientResponse
 from .helpers import call_later, set_result
 from .http import (
@@ -132,8 +132,12 @@ def _pong_not_received(self) -> None:
         if not self._closed:
             self._closed = True
             self._close_code = WSCloseCode.ABNORMAL_CLOSURE
-            self._exception = asyncio.TimeoutError()
+            self._exception = ServerTimeoutError()
             self._response.close()
+            if self._waiting and not self._closing:
+                self._reader.feed_data(
+                    WSMessage(WSMsgType.ERROR, self._exception, None)
+                )
 
     @property
     def closed(self) -> bool:

tests/test_client_ws_functional.py

@@ -6,7 +6,7 @@
 import pytest
 
 import aiohttp
-from aiohttp import hdrs, web
+from aiohttp import ServerTimeoutError, WSMsgType, hdrs, web
 from aiohttp.client_ws import ClientWSTimeout
 from aiohttp.http import WSCloseCode
 from aiohttp.pytest_plugin import AiohttpClient
@@ -685,6 +685,35 @@ async def handler(request):
     assert resp.close_code is WSCloseCode.ABNORMAL_CLOSURE
 
 
+async def test_heartbeat_no_pong_concurrent_receive(aiohttp_client: Any) -> None:
+    ping_received = False
+
+    async def handler(request):
+        nonlocal ping_received
+        ws = web.WebSocketResponse(autoping=False)
+        await ws.prepare(request)
+        msg = await ws.receive()
+        if msg.type == aiohttp.WSMsgType.PING:
+            ping_received = True
+        ws._reader.feed_eof = lambda: None
+        await asyncio.sleep(10.0)
+
+    app = web.Application()
+    app.router.add_route("GET", "/", handler)
+
+    client = await aiohttp_client(app)
+    resp = await client.ws_connect("/", heartbeat=0.1)
+    resp._reader.feed_eof = lambda: None
+
+    # Connection should be closed roughly after 1.5x heartbeat.
+    msg = await resp.receive(5.0)
+    assert ping_received
+    assert resp.close_code is WSCloseCode.ABNORMAL_CLOSURE
+    assert msg
+    assert msg.type is WSMsgType.ERROR
+    assert isinstance(msg.data, ServerTimeoutError)
+
+
 async def test_send_recv_compress(aiohttp_client: Any) -> None:
     async def handler(request):
         ws = web.WebSocketResponse()