Add new "sync once" mode for pool.connect

Fixed critical regression caused by the change in :ticket`5497` where the
connection pool "init" phase no longer occurred within mutexed isolation,
allowing other threads to proceed with the dialect uninitialized, which
could then impact the compilation of SQL statements.

This issue is essentially the same regression which was fixed many years
ago in :ticket:`2964` in dd32540dabbee0678530fb1b0868d1eb41572dca,
which was missed this time as the test suite fo
that issue only tested the pool in isolation, and assumed the
"first_connect" event would be used by the Engine.  However
:ticket:`5497` stopped using "first_connect" and no test detected
the lack of mutexing, that has been resolved here through
the addition of more tests.

This fix also identifies what is probably a bug in earlier versions
of SQLAlchemy where the "first_connect" handler would be cancelled
if the initializer failed; this is evidenced by
test_explode_in_initializer which was doing a reconnect due to
c.rollback() yet wasn't hanging.  We now solve this issue by
preventing the manufactured Connection from ever reconnecting
inside the first_connect handler.

Also remove the "_sqla_unwrap" test attribute; this is almost
not used anymore however we can use a more targeted
wrapper supplied by the testing.engines.proxying_engine
function.

See if we can also open up Oracle for "ad hoc engines" tests
now that we have better connection management logic.

Fixes: #6337
Change-Id: I4a3476625c4606f1a304dbc940d500325e8adc1a

1 files changed, 20 insertions, 9 deletions

diff --git a/lib/sqlalchemy/engine/create.py b/lib/sqlalchemy/engine/create.py
index 682d0dd5d..0351f2ebc 100644
--- a/lib/sqlalchemy/engine/create.py
+++ b/lib/sqlalchemy/engine/create.py
@@ -626,7 +626,10 @@ def create_engine(url, **kwargs):
         if k in kwargs:
             engine_args[k] = pop_kwarg(k)
 
+    # internal flags used by the test suite for instrumenting / proxying
+    # engines with mocks etc.
     _initialize = kwargs.pop("_initialize", True)
+    _wrap_do_on_connect = kwargs.pop("_wrap_do_on_connect", None)
 
     # all kwargs should be consumed
     if kwargs:
@@ -646,30 +649,38 @@ def create_engine(url, **kwargs):
     engine = engineclass(pool, dialect, u, **engine_args)
 
     if _initialize:
+
         do_on_connect = dialect.on_connect_url(url)
         if do_on_connect:
+            if _wrap_do_on_connect:
+                do_on_connect = _wrap_do_on_connect(do_on_connect)
 
             def on_connect(dbapi_connection, connection_record):
-                conn = getattr(
-                    dbapi_connection, "_sqla_unwrap", dbapi_connection
-                )
-                if conn is None:
-                    return
-
-                do_on_connect(conn)
+                do_on_connect(dbapi_connection)
 
             event.listen(pool, "connect", on_connect)
 
         def first_connect(dbapi_connection, connection_record):
             c = base.Connection(
-                engine, connection=dbapi_connection, _has_events=False
+                engine,
+                connection=dbapi_connection,
+                _has_events=False,
+                # reconnecting will be a reentrant condition, so if the
+                # connection goes away, Connection is then closed
+                _allow_revalidate=False,
             )
             c._execution_options = util.EMPTY_DICT
 
             try:
                 dialect.initialize(c)
             finally:
-                dialect.do_rollback(c.connection)
+                # note that "invalidated" and "closed" are mutually
+                # exclusive in 1.4 Connection.
+                if not c.invalidated and not c.closed:
+                    # transaction is rolled back otherwise, tested by
+                    # test/dialect/postgresql/test_dialect.py
+                    # ::MiscBackendTest::test_initial_transaction_state
+                    dialect.do_rollback(c.connection)
 
         # previously, the "first_connect" event was used here, which was then
         # scaled back if the "on_connect" handler were present.  now,