contextmanager skips rollback if trans says to skip it

Fixed issue where if an exception occurred when the :class:`_orm.Session`
were to close the connection within the :meth:`_orm.Session.commit` method,
when using a context manager for :meth:`_orm.Session.begin` , it would
attempt a rollback which would not be possible as the :class:`_orm.Session`
was in between where the transaction is committed and the connection is
then to be returned to the pool, raising the exception "this
sessiontransaction is in the committed state". This exception can occur
mostly in an asyncio context where CancelledError can be raised.

Fixes: #7388
Change-Id: I1a85a3a7eae79f3553ddf1e3d245a0d90b0a2f40
(cherry picked from commit a845da8b0fc5bb172e278c399a1de9a2e49d62af)

6 files changed, 114 insertions, 2 deletions

diff --git a/doc/build/changelog/unreleased_14/7388.rst b/doc/build/changelog/unreleased_14/7388.rst
new file mode 100644
index 000000000..1c7775a34
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/7388.rst
@@ -0,0 +1,13 @@
+.. change::
+    :tags: bug, orm
+    :tickets: 7388
+
+    Fixed issue where if an exception occurred when the :class:`_orm.Session`
+    were to close the connection within the :meth:`_orm.Session.commit` method,
+    when using a context manager for :meth:`_orm.Session.begin` , it would
+    attempt a rollback which would not be possible as the :class:`_orm.Session`
+    was in between where the transaction is committed and the connection is
+    then to be returned to the pool, raising the exception "this
+    sessiontransaction is in the committed state". This exception can occur
+    mostly in an asyncio context where CancelledError can be raised.
+
diff --git a/lib/sqlalchemy/engine/base.py b/lib/sqlalchemy/engine/base.py
index 0c27ea6d9..a5d973a2c 100644
--- a/lib/sqlalchemy/engine/base.py
+++ b/lib/sqlalchemy/engine/base.py
@@ -2371,6 +2371,13 @@ class Transaction(TransactionalContext):
     def _transaction_is_closed(self):
         return not self._deactivated_from_connection
 
+    def _rollback_can_be_called(self):
+        # for RootTransaction / NestedTransaction, it's safe to call
+        # rollback() even if the transaction is deactive and no warnings
+        # will be emitted.  tested in
+        # test_transaction.py -> test_no_rollback_in_deactive(?:_savepoint)?
+        return True
+
 
 class MarkerTransaction(Transaction):
     """A 'marker' transaction that is used for nested begin() calls.
diff --git a/lib/sqlalchemy/engine/util.py b/lib/sqlalchemy/engine/util.py
index 8eb0f1820..660ffafa0 100644
--- a/lib/sqlalchemy/engine/util.py
+++ b/lib/sqlalchemy/engine/util.py
@@ -171,6 +171,23 @@ class TransactionalContext(object):
     def _transaction_is_closed(self):
         raise NotImplementedError()
 
+    def _rollback_can_be_called(self):
+        """indicates the object is in a state that is known to be acceptable
+        for rollback() to be called.
+
+        This does not necessarily mean rollback() will succeed or not raise
+        an error, just that there is currently no state detected that indicates
+        rollback() would fail or emit warnings.
+
+        It also does not mean that there's a transaction in progress, as
+        it is usually safe to call rollback() even if no transaction is
+        present.
+
+        .. versionadded:: 1.4.28
+
+        """
+        raise NotImplementedError()
+
     def _get_subject(self):
         raise NotImplementedError()
 
@@ -216,7 +233,8 @@ class TransactionalContext(object):
                 self.commit()
             except:
                 with util.safe_reraise():
-                    self.rollback()
+                    if self._rollback_can_be_called():
+                        self.rollback()
             finally:
                 if not out_of_band_exit:
                     subject._trans_context_manager = self._outer_trans_ctx
@@ -227,7 +245,8 @@ class TransactionalContext(object):
                     if not self._transaction_is_closed():
                         self.close()
                 else:
-                    self.rollback()
+                    if self._rollback_can_be_called():
+                        self.rollback()
             finally:
                 if not out_of_band_exit:
                     subject._trans_context_manager = self._outer_trans_ctx
diff --git a/lib/sqlalchemy/orm/session.py b/lib/sqlalchemy/orm/session.py
index bb12f7021..034651326 100644
--- a/lib/sqlalchemy/orm/session.py
+++ b/lib/sqlalchemy/orm/session.py
@@ -939,6 +939,9 @@ class SessionTransaction(TransactionalContext):
     def _transaction_is_closed(self):
         return self._state is CLOSED
 
+    def _rollback_can_be_called(self):
+        return self._state not in (COMMITTED, CLOSED)
+
 
 class Session(_SessionClassMethods):
     """Manages persistence operations for ORM-mapped objects.
diff --git a/test/engine/test_transaction.py b/test/engine/test_transaction.py
index 9e6142022..43b42647e 100644
--- a/test/engine/test_transaction.py
+++ b/test/engine/test_transaction.py
@@ -487,6 +487,36 @@ class TransactionTest(fixtures.TablesTest):
         result = connection.exec_driver_sql("select * from users")
         assert len(result.fetchall()) == 0
 
+    @testing.requires.independent_connections
+    def test_no_rollback_in_deactive(self, local_connection):
+        """test #7388"""
+
+        def fail(*arg, **kw):
+            raise BaseException("some base exception")
+
+        with mock.patch.object(testing.db.dialect, "do_commit", fail):
+            with expect_raises_message(BaseException, "some base exception"):
+                with local_connection.begin():
+                    pass
+
+    @testing.requires.independent_connections
+    @testing.requires.savepoints
+    def test_no_rollback_in_deactive_savepoint(self, local_connection):
+        """test #7388"""
+
+        def fail(*arg, **kw):
+            raise BaseException("some base exception")
+
+        with mock.patch.object(
+            testing.db.dialect, "do_release_savepoint", fail
+        ):
+            with local_connection.begin():
+                with expect_raises_message(
+                    BaseException, "some base exception"
+                ):
+                    with local_connection.begin_nested():
+                        pass
+
     @testing.requires.savepoints
     def test_nested_subtransaction_rollback(self, local_connection):
         connection = local_connection
diff --git a/test/orm/test_transaction.py b/test/orm/test_transaction.py
index f0ef37230..603ec079a 100644
--- a/test/orm/test_transaction.py
+++ b/test/orm/test_transaction.py
@@ -513,6 +513,46 @@ class SessionTransactionTest(fixtures.RemovesEvents, FixtureTest):
         assert conn.closed
         assert not fairy.is_valid
 
+    @testing.requires.independent_connections
+    def test_no_rollback_in_committed_state(self):
+        """test #7388
+
+        Prior to the fix, using the session.begin() context manager
+        would produce the error "This session is in 'committed' state; no
+        further SQL can be emitted ", when it attempted to call .rollback()
+        if the connection.close() operation failed inside of session.commit().
+
+        While the real exception was chained inside, this still proved to
+        be misleading so we now skip the rollback() in this specific case
+        and allow the original error to be raised.
+
+        """
+
+        sess = fixture_session()
+
+        def fail(*arg, **kw):
+            raise BaseException("some base exception")
+
+        with mock.patch.object(
+            testing.db.dialect, "do_rollback", side_effect=fail
+        ) as fail_mock, mock.patch.object(
+            testing.db.dialect,
+            "do_commit",
+            side_effect=testing.db.dialect.do_commit,
+        ) as succeed_mock:
+
+            # sess.begin() -> commit().  why would do_rollback() be called?
+            # because of connection pool finalize_fairy *after* the commit.
+            # this will cause the conn.close() in session.commit() to fail,
+            # but after the DB commit succeeded.
+            with expect_raises_message(BaseException, "some base exception"):
+                with sess.begin():
+                    conn = sess.connection()
+                    fairy_conn = conn.connection
+
+        eq_(succeed_mock.mock_calls, [mock.call(fairy_conn)])
+        eq_(fail_mock.mock_calls, [mock.call(fairy_conn)])
+
     def test_continue_flushing_on_commit(self):
         """test that post-flush actions get flushed also if
         we're in commit()"""