- Fixed bug where a "branched" connection, that is the kind you get

when you call :meth:`.Connection.connect`, would not share invalidation
status with the parent.  The architecture of branching has been tweaked
a bit so that the branched connection defers to the parent for
all invalidation status and operations.
fixes #3215

3 files changed, 74 insertions, 11 deletions

diff --git a/doc/build/changelog/changelog_10.rst b/doc/build/changelog/changelog_10.rst
index 536288c8f..a4f3dd6e5 100644
--- a/doc/build/changelog/changelog_10.rst
+++ b/doc/build/changelog/changelog_10.rst
@@ -22,6 +22,16 @@
     on compatibility concerns, see :doc:`/changelog/migration_10`.
 
     .. change::
+        :tags: bug, sql, engine
+        :tickets: 3215
+
+        Fixed bug where a "branched" connection, that is the kind you get
+        when you call :meth:`.Connection.connect`, would not share invalidation
+        status with the parent.  The architecture of branching has been tweaked
+        a bit so that the branched connection defers to the parent for
+        all invalidation status and operations.
+
+    .. change::
         :tags: bug, declarative
         :tickets: 2670
 
diff --git a/lib/sqlalchemy/engine/base.py b/lib/sqlalchemy/engine/base.py
index d2cc8890f..ec7aed1c3 100644
--- a/lib/sqlalchemy/engine/base.py
+++ b/lib/sqlalchemy/engine/base.py
@@ -45,7 +45,7 @@ class Connection(Connectable):
     """
 
     def __init__(self, engine, connection=None, close_with_result=False,
-                 _branch=False, _execution_options=None,
+                 _branch_from=None, _execution_options=None,
                  _dispatch=None,
                  _has_events=None):
         """Construct a new Connection.
@@ -61,7 +61,8 @@ class Connection(Connectable):
         self.__transaction = None
         self.should_close_with_result = close_with_result
         self.__savepoint_seq = 0
-        self.__branch = _branch
+        self.__branch_from = _branch_from
+        self.__branch = _branch_from is not None
         self.__invalid = False
         self.__can_reconnect = True
         if _dispatch:
@@ -82,7 +83,7 @@ class Connection(Connectable):
             self._execution_options = engine._execution_options
 
         if self._has_events or self.engine._has_events:
-            self.dispatch.engine_connect(self, _branch)
+            self.dispatch.engine_connect(self, self.__branch)
 
     def _branch(self):
         """Return a new Connection which references this Connection's
@@ -92,13 +93,26 @@ class Connection(Connectable):
         This is used to execute "sub" statements within a single execution,
         usually an INSERT statement.
         """
+        if self.__branch_from:
+            return self.__branch_from._branch()
+        else:
+            return self.engine._connection_cls(
+                self.engine,
+                self.__connection,
+                _branch_from=self,
+                _has_events=self._has_events,
+                _dispatch=self.dispatch)
+
+    @property
+    def _root(self):
+        """return the 'root' connection.
 
-        return self.engine._connection_cls(
-            self.engine,
-            self.__connection,
-            _branch=True,
-            _has_events=self._has_events,
-            _dispatch=self.dispatch)
+        Returns 'self' if this connection is not a branch, else
+        returns the root connection from which we ultimately branched."""
+        if self.__branch_from:
+            return self.__branch_from
+        else:
+            return self
 
     def _clone(self):
         """Create a shallow copy of this Connection.
@@ -218,13 +232,13 @@ class Connection(Connectable):
         """Return True if this connection is closed."""
 
         return '_Connection__connection' not in self.__dict__ \
-            and not self.__can_reconnect
+            and not self._root.__can_reconnect
 
     @property
     def invalidated(self):
         """Return True if this connection was invalidated."""
 
-        return self.__invalid
+        return self._root.__invalid
 
     @property
     def connection(self):
@@ -236,6 +250,9 @@ class Connection(Connectable):
             return self._revalidate_connection()
 
     def _revalidate_connection(self):
+        if self.__branch_from:
+            return self._root._revalidate_connection()
+
         if self.__can_reconnect and self.__invalid:
             if self.__transaction is not None:
                 raise exc.InvalidRequestError(
@@ -343,6 +360,10 @@ class Connection(Connectable):
             :ref:`pool_connection_invalidation`
 
         """
+        if self.__branch_from:
+            self._root.invalidate()
+            return
+
         if self.invalidated:
             return
 
diff --git a/test/engine/test_reconnect.py b/test/engine/test_reconnect.py
index c82cca5a1..26a607301 100644
--- a/test/engine/test_reconnect.py
+++ b/test/engine/test_reconnect.py
@@ -504,6 +504,38 @@ class RealReconnectTest(fixtures.TestBase):
         # pool isn't replaced
         assert self.engine.pool is p2
 
+    def test_branched_invalidate_branch_to_parent(self):
+        c1 = self.engine.connect()
+
+        c1_branch = c1.connect()
+        eq_(c1_branch.execute(select([1])).scalar(), 1)
+
+        self.engine.test_shutdown()
+
+        _assert_invalidated(c1_branch.execute, select([1]))
+        assert c1.invalidated
+        assert c1_branch.invalidated
+
+        c1_branch._revalidate_connection()
+        assert not c1.invalidated
+        assert not c1_branch.invalidated
+
+    def test_branched_invalidate_parent_to_branch(self):
+        c1 = self.engine.connect()
+
+        c1_branch = c1.connect()
+        eq_(c1_branch.execute(select([1])).scalar(), 1)
+
+        self.engine.test_shutdown()
+
+        _assert_invalidated(c1.execute, select([1]))
+        assert c1.invalidated
+        assert c1_branch.invalidated
+
+        c1._revalidate_connection()
+        assert not c1.invalidated
+        assert not c1_branch.invalidated
+
     def test_ensure_is_disconnect_gets_connection(self):
         def is_disconnect(e, conn, cursor):
             # connection is still present