Invalidate on failed connect handler

Fixed bug in connection pool where a connection could be present in the
pool without all of its "connect" event handlers called, if a previous
"connect" handler threw an exception; note that the dialects themselves
have connect handlers that emit SQL, such as those which set transaction
isolation, which can fail if the database is in a non-available state, but
still allows a connection.  The connection is now invalidated first if any
of the connect handlers fail.

Change-Id: I61d6f4827a98ab8455f1c3e1c55d046eeccec09a
Fixes: #4225

3 files changed, 49 insertions, 4 deletions

diff --git a/doc/build/changelog/unreleased_12/4225.rst b/doc/build/changelog/unreleased_12/4225.rst
new file mode 100644
index 000000000..99b0a6a7a
--- /dev/null
+++ b/doc/build/changelog/unreleased_12/4225.rst
@@ -0,0 +1,12 @@
+.. change::
+    :tags: bug, engine
+    :tickets: 4225
+    :versions: 1.3.0b1
+
+    Fixed bug in connection pool where a connection could be present in the
+    pool without all of its "connect" event handlers called, if a previous
+    "connect" handler threw an exception; note that the dialects themselves
+    have connect handlers that emit SQL, such as those which set transaction
+    isolation, which can fail if the database is in a non-available state, but
+    still allows a connection.  The connection is now invalidated first if any
+    of the connect handlers fail.
diff --git a/lib/sqlalchemy/pool.py b/lib/sqlalchemy/pool.py
index 102d50c18..89a4cea7c 100644
--- a/lib/sqlalchemy/pool.py
+++ b/lib/sqlalchemy/pool.py
@@ -532,9 +532,9 @@ class _ConnectionRecord(object):
         rec = pool._do_get()
         try:
             dbapi_connection = rec.get_connection()
-        except:
+        except Exception as err:
             with util.safe_reraise():
-                rec.checkin()
+                rec._checkin_failed(err)
         echo = pool._should_log_debug()
         fairy = _ConnectionFairy(dbapi_connection, rec, echo)
         rec.fairy_ref = weakref.ref(
@@ -550,6 +550,10 @@ class _ConnectionRecord(object):
                               dbapi_connection)
         return fairy
 
+    def _checkin_failed(self, err):
+        self.invalidate(e=err)
+        self.checkin()
+
     def checkin(self):
         self.fairy_ref = None
         connection = self.connection
@@ -840,9 +844,9 @@ class _ConnectionFairy(object):
                 try:
                     fairy.connection = \
                         fairy._connection_record.get_connection()
-                except:
+                except Exception as err:
                     with util.safe_reraise():
-                        fairy._connection_record.checkin()
+                        fairy._connection_record._checkin_failed(err)
 
                 attempts -= 1
 
diff --git a/test/engine/test_pool.py b/test/engine/test_pool.py
index e3f34dfa7..b699afbc6 100644
--- a/test/engine/test_pool.py
+++ b/test/engine/test_pool.py
@@ -730,6 +730,35 @@ class PoolEventsTest(PoolTestBase):
         p2.connect()
         eq_(canary, ["listen_one", "listen_two", "listen_one", "listen_three"])
 
+    def test_connect_event_fails_invalidates(self):
+        fail = False
+
+        def listen_one(conn, rec):
+            if fail:
+                raise Exception("it failed")
+
+        def listen_two(conn, rec):
+            rec.info['important_flag'] = True
+
+        p1 = pool.QueuePool(
+            creator=MockDBAPI().connect, pool_size=1, max_overflow=0)
+        event.listen(p1, 'connect', listen_one)
+        event.listen(p1, 'connect', listen_two)
+
+        conn = p1.connect()
+        eq_(conn.info['important_flag'], True)
+        conn.invalidate()
+        conn.close()
+
+        fail = True
+        assert_raises(Exception, p1.connect)
+
+        fail = False
+
+        conn = p1.connect()
+        eq_(conn.info['important_flag'], True)
+        conn.close()
+
     def teardown(self):
         # TODO: need to get remove() functionality
         # going