run handle error for commit/rollback fail and cancel transaction

Fixed bug in asyncpg dialect where a failure during a "commit" or less
likely a "rollback" should cancel the entire transaction; it's no longer
possible to emit rollback. Previously the connection would continue to
await a rollback that could not succeed as asyncpg would reject it.

Fixes: #5824
Change-Id: I5a4916740c269b410f4d1a78ed25191de344b9d0

1 files changed, 18 insertions, 6 deletions

diff --git a/lib/sqlalchemy/dialects/postgresql/asyncpg.py b/lib/sqlalchemy/dialects/postgresql/asyncpg.py
index e542c77f4..424ed0d50 100644
--- a/lib/sqlalchemy/dialects/postgresql/asyncpg.py
+++ b/lib/sqlalchemy/dialects/postgresql/asyncpg.py
@@ -615,6 +615,10 @@ class AsyncAdapt_asyncpg_connection:
         return prepared_stmt, attributes
 
     def _handle_exception(self, error):
+        if self._connection.is_closed():
+            self._transaction = None
+            self._started = False
+
         if not isinstance(error, AsyncAdapt_asyncpg_dbapi.Error):
             exception_mapping = self.dbapi._asyncpg_error_translate
 
@@ -669,15 +673,23 @@ class AsyncAdapt_asyncpg_connection:
 
     def rollback(self):
         if self._started:
-            self.await_(self._transaction.rollback())
-            self._transaction = None
-            self._started = False
+            try:
+                self.await_(self._transaction.rollback())
+            except Exception as error:
+                self._handle_exception(error)
+            finally:
+                self._transaction = None
+                self._started = False
 
     def commit(self):
         if self._started:
-            self.await_(self._transaction.commit())
-            self._transaction = None
-            self._started = False
+            try:
+                self.await_(self._transaction.commit())
+            except Exception as error:
+                self._handle_exception(error)
+            finally:
+                self._transaction = None
+                self._started = False
 
     def close(self):
         self.rollback()