fix closed MySQL 8.0 connections not always being detected properly

ensure connections are closed properly when the server connection was lost

Author: Nothing4You

CHANGES.txt

@@ -1,6 +1,14 @@
 Changes
 -------
 
+pending release
+^^^^^^^^^^^^^^^^^^^
+
+* Fix timed out MySQL 8.0 connections raising InternalError rather than OperationalError #660
+* Fix timed out MySQL 8.0 connections being returned from Pool #660
+* Ensure connections are properly closed before raising OperationalError when the server connection is lost #660
+
+
 0.0.22 (2021-11-14)
 ^^^^^^^^^^^^^^^^^^^
 

aiomysql/connection.py

@@ -15,6 +15,7 @@
 from pymysql.constants import SERVER_STATUS
 from pymysql.constants import CLIENT
 from pymysql.constants import COMMAND
+from pymysql.constants import CR
 from pymysql.constants import FIELD_TYPE
 from pymysql.util import byte2int, int2byte
 from pymysql.converters import (escape_item, encoders, decoders,
@@ -470,7 +471,7 @@ async def set_charset(self, charset):
 
     async def _connect(self):
         # TODO: Set close callback
-        # raise OperationalError(2006,
+        # raise OperationalError(CR.CR_SERVER_GONE_ERROR,
         # "MySQL server has gone away (%r)" % (e,))
         try:
             if self._unix_socket and self._host in ('localhost', '127.0.0.1'):
@@ -569,6 +570,13 @@ async def _read_packet(self, packet_type=MysqlPacket):
             # we increment in both write_packet and read_packet. The count
             # is reset at new COMMAND PHASE.
             if packet_number != self._next_seq_id:
+                if packet_number == 0:
+                    # MySQL 8.0 sends error packet with seqno==0 when shutdown
+                    self.close()
+                    raise OperationalError(
+                        CR.CR_SERVER_LOST,
+                        "Lost connection to MySQL server during query")
+
                 raise InternalError(
                     "Packet sequence number wrong - got %d expected %d" %
                     (packet_number, self._next_seq_id))
@@ -596,10 +604,12 @@ async def _read_bytes(self, num_bytes):
             data = await self._reader.readexactly(num_bytes)
         except asyncio.IncompleteReadError as e:
             msg = "Lost connection to MySQL server during query"
-            raise OperationalError(2013, msg) from e
+            self.close()
+            raise OperationalError(CR.CR_SERVER_LOST, msg) from e
         except (IOError, OSError) as e:
             msg = "Lost connection to MySQL server during query (%s)" % (e,)
-            raise OperationalError(2013, msg) from e
+            self.close()
+            raise OperationalError(CR.CR_SERVER_LOST, msg) from e
         return data
 
     def _write_bytes(self, data):

aiomysql/pool.py

@@ -143,7 +143,7 @@ async def _acquire(self):
                     await self._cond.wait()
 
     async def _fill_free_pool(self, override_min):
-        # iterate over free connections and remove timeouted ones
+        # iterate over free connections and remove timed out ones
         free_size = len(self._free)
         n = 0
         while n < free_size:
@@ -152,6 +152,11 @@ async def _fill_free_pool(self, override_min):
                 self._free.pop()
                 conn.close()
 
+            # TODO: fixme, we should not use internal attributes
+            elif conn._reader._eof:
+                self._free.pop()
+                conn.close()
+
             elif (self._recycle > -1 and
                   self._loop.time() - conn.last_usage > self._recycle):
                 self._free.pop()