Add new "exec_once_unless_exception" system; apply to dialect.initialize

Fixed an issue whereby if the dialect "initialize" process which occurs on
first connect would encounter an unexpected exception, the initialize
process would fail to complete and then no longer attempt on subsequent
connection attempts, leaving the dialect in an un-initialized, or partially
initialized state, within the scope of parameters that need to be
established based on inspection of a live connection.   The "invoke once"
logic in the event system has been reworked to accommodate for this
occurrence using new, private API features that establish an "exec once"
hook that will continue to allow the initializer to fire off on subsequent
connections, until it completes without raising an exception. This does not
impact the behavior of the existing ``once=True`` flag within the event
system.

Fixes: #4807
Change-Id: Iec32999b61b6af4b38b6719e0c2651454619078c

5 files changed, 50 insertions, 15 deletions

diff --git a/lib/sqlalchemy/engine/create.py b/lib/sqlalchemy/engine/create.py
index cc8304131..72be6009b 100644
--- a/lib/sqlalchemy/engine/create.py
+++ b/lib/sqlalchemy/engine/create.py
@@ -529,7 +529,9 @@ def create_engine(url, **kwargs):
             dialect.initialize(c)
             dialect.do_rollback(c.connection)
 
-        event.listen(pool, "first_connect", first_connect, once=True)
+        event.listen(
+            pool, "first_connect", first_connect, _once_unless_exception=True
+        )
 
     dialect_cls.engine_created(engine)
     if entrypoint is not dialect_cls:
diff --git a/lib/sqlalchemy/event/attr.py b/lib/sqlalchemy/event/attr.py
index 9dfa89809..b6c48fa6c 100644
--- a/lib/sqlalchemy/event/attr.py
+++ b/lib/sqlalchemy/event/attr.py
@@ -250,7 +250,9 @@ class _EmptyListener(_InstanceLevelDispatch):
     def _needs_modify(self, *args, **kw):
         raise NotImplementedError("need to call for_modify()")
 
-    exec_once = insert = append = remove = clear = _needs_modify
+    exec_once = (
+        exec_once_unless_exception
+    ) = insert = append = remove = clear = _needs_modify
 
     def __call__(self, *args, **kw):
         """Execute this event."""
@@ -276,17 +278,40 @@ class _CompoundListener(_InstanceLevelDispatch):
     def _memoized_attr__exec_once_mutex(self):
         return threading.Lock()
 
+    def _exec_once_impl(self, retry_on_exception, *args, **kw):
+        with self._exec_once_mutex:
+            if not self._exec_once:
+                try:
+                    self(*args, **kw)
+                    exception = False
+                except:
+                    exception = True
+                    raise
+                finally:
+                    if not exception or not retry_on_exception:
+                        self._exec_once = True
+
     def exec_once(self, *args, **kw):
         """Execute this event, but only if it has not been
         executed already for this collection."""
 
         if not self._exec_once:
-            with self._exec_once_mutex:
-                if not self._exec_once:
-                    try:
-                        self(*args, **kw)
-                    finally:
-                        self._exec_once = True
+            self._exec_once_impl(False, *args, **kw)
+
+    def exec_once_unless_exception(self, *args, **kw):
+        """Execute this event, but only if it has not been
+        executed already for this collection, or was called
+        by a previous exec_once_unless_exception call and
+        raised an exception.
+
+        If exec_once was already called, then this method will never run
+        the callable regardless of whether it raised or not.
+
+        .. versionadded:: 1.3.8
+
+        """
+        if not self._exec_once:
+            self._exec_once_impl(True, *args, **kw)
 
     def __call__(self, *args, **kw):
         """Execute this event."""
diff --git a/lib/sqlalchemy/event/registry.py b/lib/sqlalchemy/event/registry.py
index 07b961c01..2b8619b6e 100644
--- a/lib/sqlalchemy/event/registry.py
+++ b/lib/sqlalchemy/event/registry.py
@@ -192,6 +192,7 @@ class _EventKey(object):
 
     def listen(self, *args, **kw):
         once = kw.pop("once", False)
+        once_unless_exception = kw.pop("_once_unless_exception", False)
         named = kw.pop("named", False)
 
         target, identifier, fn = (
@@ -212,10 +213,12 @@ class _EventKey(object):
         if hasattr(stub_function, "_sa_warn"):
             stub_function._sa_warn()
 
-        if once:
-            self.with_wrapper(util.only_once(self._listen_fn)).listen(
-                *args, **kw
-            )
+        if once or once_unless_exception:
+            self.with_wrapper(
+                util.only_once(
+                    self._listen_fn, retry_on_exception=once_unless_exception
+                )
+            ).listen(*args, **kw)
         else:
             self.dispatch_target.dispatch._listen(self, *args, **kw)
 
diff --git a/lib/sqlalchemy/pool/base.py b/lib/sqlalchemy/pool/base.py
index 2325e7faa..c45f836db 100644
--- a/lib/sqlalchemy/pool/base.py
+++ b/lib/sqlalchemy/pool/base.py
@@ -604,7 +604,7 @@ class _ConnectionRecord(object):
             if first_connect_check:
                 pool.dispatch.first_connect.for_modify(
                     pool.dispatch
-                ).exec_once(self.connection, self)
+                ).exec_once_unless_exception(self.connection, self)
             if pool.dispatch.connect:
                 pool.dispatch.connect(self.connection, self)
 
diff --git a/lib/sqlalchemy/util/langhelpers.py b/lib/sqlalchemy/util/langhelpers.py
index 12fc5c0e8..f3f3f9ea5 100644
--- a/lib/sqlalchemy/util/langhelpers.py
+++ b/lib/sqlalchemy/util/langhelpers.py
@@ -1487,7 +1487,7 @@ def warn_limited(msg, args):
     warnings.warn(msg, exc.SAWarning, stacklevel=2)
 
 
-def only_once(fn):
+def only_once(fn, retry_on_exception):
     """Decorate the given function to be a no-op after it is called exactly
     once."""
 
@@ -1499,7 +1499,12 @@ def only_once(fn):
         strong_fn = fn  # noqa
         if once:
             once_fn = once.pop()
-            return once_fn(*arg, **kw)
+            try:
+                return once_fn(*arg, **kw)
+            except:
+                if retry_on_exception:
+                    once.insert(0, once_fn)
+                raise
 
     return go