Merge "contextmanager skips rollback if trans says to skip it" into main

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 5419a912e..f75f2ac9f 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()"""