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

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 fbd8fe7df..b11ffd871 100644
--- a/lib/sqlalchemy/engine/base.py
+++ b/lib/sqlalchemy/engine/base.py
@@ -2022,6 +2022,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 RootTransaction(Transaction):
     """Represent the "root" transaction on a :class:`_engine.Connection`.
diff --git a/lib/sqlalchemy/engine/util.py b/lib/sqlalchemy/engine/util.py
index 732ae7fa1..e88b9ebf3 100644
--- a/lib/sqlalchemy/engine/util.py
+++ b/lib/sqlalchemy/engine/util.py
@@ -98,6 +98,23 @@ class TransactionalContext:
     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()
 
@@ -143,7 +160,8 @@ class TransactionalContext:
                 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
@@ -154,7 +172,8 @@ class TransactionalContext:
                     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 8212a111d..b6fc99baf 100644
--- a/lib/sqlalchemy/orm/session.py
+++ b/lib/sqlalchemy/orm/session.py
@@ -940,6 +940,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 d47fa38c2..696391512 100644
--- a/test/engine/test_transaction.py
+++ b/test/engine/test_transaction.py
@@ -373,6 +373,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 3574c4507..c5d06ba88 100644
--- a/test/orm/test_transaction.py
+++ b/test/orm/test_transaction.py
@@ -512,6 +512,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()"""