Handle ConnectionClosed exception in keepalive_ping.

Fix #551.

Thanks @Harmon758 for diagnosing the root cause and proposing a fix.

Author: aaugustin

docs/changelog.rst

@@ -59,6 +59,9 @@ Also:
 * Improved support for sending fragmented messages by accepting asynchronous
   iterators in :meth:`~protocol.WebSocketCommonProtocol.send`.
 
+* Prevented spurious log messages about :exc:`~exceptions.ConnectionClosed`
+  exceptions in keepalive ping task.
+
 * Avoided a crash of a ``extra_headers`` callable returns ``None``.
 
 * Enabled readline in the interactive client.

src/websockets/protocol.py

@@ -980,6 +980,7 @@ async def keepalive_ping(self) -> None:
 
         This coroutine exits when the connection terminates and one of the
         following happens:
+
         - :meth:`ping` raises :exc:`ConnectionClosed`, or
         - :meth:`close_connection` cancels :attr:`keepalive_ping_task`.
 
@@ -991,11 +992,12 @@ async def keepalive_ping(self) -> None:
             while True:
                 await asyncio.sleep(self.ping_interval, loop=self.loop)
 
-                # ping() cannot raise ConnectionClosed, only CancelledError:
-                # - If the connection is CLOSING, keepalive_ping_task will be
-                #   canceled by close_connection() before ping() returns.
-                # - If the connection is CLOSED, keepalive_ping_task must be
-                #   canceled already.
+                # ping() raises CancelledError if the connection is closed,
+                # when close_connection() cancels self.keepalive_ping_task.
+
+                # ping() raises ConnectionClosed if the connection is lost,
+                # when connection_lost() calls abort_keepalive_pings().
+
                 ping_waiter = await self.ping()
 
                 if self.ping_timeout is not None:
@@ -1011,6 +1013,9 @@ async def keepalive_ping(self) -> None:
         except asyncio.CancelledError:
             raise
 
+        except ConnectionClosed:
+            pass
+
         except Exception:
             logger.warning("Unexpected exception in keepalive ping task", exc_info=True)
 

tests/test_protocol.py

@@ -1074,6 +1074,31 @@ def test_keepalive_ping_stops_when_connection_closed(self):
         # The keepalive ping task terminated.
         self.assertTrue(self.protocol.keepalive_ping_task.cancelled())
 
+    def test_keepalive_ping_does_not_crash_when_connection_lost(self):
+        self.restart_protocol_with_keepalive_ping()
+        # Clog incoming queue. This lets connection_lost() abort pending pings
+        # with a ConnectionClosed exception before transfer_data_task
+        # terminates and close_connection cancels keepalive_ping_task.
+        self.protocol.max_queue = 1
+        self.receive_frame(Frame(True, OP_TEXT, b"1"))
+        self.receive_frame(Frame(True, OP_TEXT, b"2"))
+        # Ping is sent at 3ms.
+        self.loop.run_until_complete(asyncio.sleep(4 * MS))
+        ping_waiter, = tuple(self.protocol.pings.values())
+        # Connection drops.
+        self.receive_eof()
+        self.loop.run_until_complete(self.protocol.wait_closed())
+
+        # The ping waiter receives a ConnectionClosed exception.
+        with self.assertRaises(ConnectionClosed):
+            ping_waiter.result()
+        # The keepalive ping task terminated properly.
+        self.assertIsNone(self.protocol.keepalive_ping_task.result())
+
+        # Unclog incoming queue to terminate the test quickly.
+        self.loop.run_until_complete(self.protocol.recv())
+        self.loop.run_until_complete(self.protocol.recv())
+
     def test_keepalive_ping_with_no_ping_interval(self):
         self.restart_protocol_with_keepalive_ping(ping_interval=None)