catch all BaseException in pool and revert failed checkouts

Fixed a long-standing race condition in the connection pool which could
occur under eventlet/gevent monkeypatching schemes in conjunction with the
use of eventlet/gevent ``Timeout`` conditions, where a connection pool
checkout that's interrupted due to the timeout would fail to clean up the
failed state, causing the underlying connection record and sometimes the
database connection itself to "leak", leaving the pool in an invalid state
with unreachable entries. This issue was first identified and fixed in
SQLAlchemy 1.2 for :ticket:`4225`, however the failure modes detected in
that fix failed to accommodate for ``BaseException``, rather than
``Exception``, which prevented eventlet/gevent ``Timeout`` from being
caught. In addition, a block within initial pool connect has also been
identified and hardened with a ``BaseException`` -> "clean failed connect"
block to accommodate for the same condition in this location.
Big thanks to Github user @niklaus for their tenacious efforts in
identifying and describing this intricate issue.

Fixes: #8974
Change-Id: I95a0e1f080d0cee6f1a66977432a586fdf87f686

1 files changed, 25 insertions, 5 deletions

diff --git a/lib/sqlalchemy/pool/base.py b/lib/sqlalchemy/pool/base.py
index 47c39791c..7b211afd9 100644
--- a/lib/sqlalchemy/pool/base.py
+++ b/lib/sqlalchemy/pool/base.py
@@ -380,10 +380,12 @@ class Pool(log.Identified, event.EventTarget):
                 self._dialect.do_terminate(connection)
             else:
                 self._dialect.do_close(connection)
-        except Exception:
+        except BaseException as e:
             self.logger.error(
                 "Exception closing connection %r", connection, exc_info=True
             )
+            if not isinstance(e, Exception):
+                raise
 
     def _create_connection(self) -> ConnectionPoolEntry:
         """Called by subclasses to create a new ConnectionRecord."""
@@ -714,9 +716,11 @@ class _ConnectionRecord(ConnectionPoolEntry):
 
         try:
             dbapi_connection = rec.get_connection()
-        except Exception as err:
+        except BaseException as err:
             with util.safe_reraise():
                 rec._checkin_failed(err, _fairy_was_created=False)
+
+            # not reached, for code linters only
             raise
 
         echo = pool._should_log_debug()
@@ -738,7 +742,7 @@ class _ConnectionRecord(ConnectionPoolEntry):
         return fairy
 
     def _checkin_failed(
-        self, err: Exception, _fairy_was_created: bool = True
+        self, err: BaseException, _fairy_was_created: bool = True
     ) -> None:
         self.invalidate(e=err)
         self.checkin(
@@ -893,7 +897,7 @@ class _ConnectionRecord(ConnectionPoolEntry):
             self.dbapi_connection = connection = pool._invoke_creator(self)
             pool.logger.debug("Created new connection %r", connection)
             self.fresh = True
-        except Exception as e:
+        except BaseException as e:
             with util.safe_reraise():
                 pool.logger.debug("Error on connect(): %s", e)
         else:
@@ -1271,6 +1275,7 @@ class _ConnectionFairy(PoolProxiedConnection):
         # here.
 
         attempts = 2
+
         while attempts > 0:
             connection_is_fresh = fairy._connection_record.fresh
             fairy._connection_record.fresh = False
@@ -1323,7 +1328,7 @@ class _ConnectionFairy(PoolProxiedConnection):
                     fairy.dbapi_connection = (
                         fairy._connection_record.get_connection()
                     )
-                except Exception as err:
+                except BaseException as err:
                     with util.safe_reraise():
                         fairy._connection_record._checkin_failed(
                             err,
@@ -1341,6 +1346,21 @@ class _ConnectionFairy(PoolProxiedConnection):
                     raise
 
                 attempts -= 1
+            except BaseException as be_outer:
+                with util.safe_reraise():
+                    rec = fairy._connection_record
+                    if rec is not None:
+                        rec._checkin_failed(
+                            be_outer,
+                            _fairy_was_created=True,
+                        )
+
+                    # prevent _ConnectionFairy from being carried
+                    # in the stack trace, see above
+                    del fairy
+
+                # never called, this is for code linters
+                raise
 
         pool.logger.info("Reconnection attempts exhausted on checkout")
         fairy.invalidate()