Fix connection left in broken state when BEGIN fails or ROLLBACK errors

.gitignore

@@ -24,4 +24,7 @@ doc/api/
 *.iws
 
 # VS Code IDE
-.vscode/
+.vscode/
+
+# Reference libraries for bug research
+_reference/

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## 3.5.12
+
+- Fix `runTx` silently rolling back after `ROLLBACK TO SAVEPOINT` recovery: clear stale `_transactionException` when PostgreSQL confirms a healthy transaction state.
+- Fix connection permanently blocked when `BEGIN` fails inside `runTx` (stale `_activeTransaction` state).
+- Fix connection left in undefined PostgreSQL state when `ROLLBACK` fails after a transaction error.
+
 ## 3.5.11
 
 - Adding `JsonbListView` with `isSqlNull(int index)` method to check if a JSONB array has SQL or JSON null value.

lib/src/v3/connection.dart

@@ -533,6 +533,13 @@ class PgConnectionImplementation extends _PgSessionBase implements Connection {
           _pending!.handleError(exception);
         }
       } else if (_pending != null) {
+        // If PostgreSQL reports the transaction is healthy (e.g. after a
+        // successful ROLLBACK TO SAVEPOINT), clear any stale exception so that
+        // mayCommit can return true and the outer runTx can COMMIT.
+        if (message is ReadyForQueryMessage &&
+            message.state == ReadyForQueryMessageState.transaction) {
+          _activeTransaction?._transactionException = null;
+        }
         await _pending!.handleMessage(message);
       }
     } finally {
@@ -588,14 +595,22 @@ class PgConnectionImplementation extends _PgSessionBase implements Connection {
         beginQuery = 'BEGIN;';
       }
 
-      await transaction.execute(Sql(beginQuery), queryMode: QueryMode.simple);
-
       try {
+        await transaction.execute(Sql(beginQuery), queryMode: QueryMode.simple);
+
         final result = await fn(transaction);
         if (transaction.mayCommit) {
           await transaction._sendAndMarkClosed('COMMIT;');
         } else if (!transaction._sessionClosed) {
-          await transaction._sendAndMarkClosed('ROLLBACK;');
+          try {
+            await transaction._sendAndMarkClosed('ROLLBACK;');
+          } catch (rollbackEx) {
+            // ROLLBACK failed — PG connection state is undefined, close it.
+            _connection._closeAfterError(
+              rollbackEx is PgException ? rollbackEx : null,
+            );
+            rethrow;
+          }
         }
 
         // If we have received an error while the transaction was active, it
@@ -609,10 +624,13 @@ class PgConnectionImplementation extends _PgSessionBase implements Connection {
         if (!transaction._sessionClosed) {
           try {
             await transaction._sendAndMarkClosed('ROLLBACK;');
-          } catch (_) {
-            // checking the outer exception
+          } catch (rollbackEx) {
+            // ROLLBACK failed — PG connection state is undefined, close it.
+            _connection._closeAfterError(
+              rollbackEx is PgException ? rollbackEx : null,
+            );
             if (e is PgException) {
-              // Ignore exception of rollback, as the earlier exception takes precedence.
+              // Original exception takes precedence over rollback failure.
             } else {
               // Do not ignore the exception here, it may be an implementation bug we are swallowing.
               rethrow;
@@ -1227,8 +1245,21 @@ class _TransactionSession extends _PgSessionBase implements TxSession {
         _connection._activeTransaction = null;
       },
     );
-    await querySubscription.asFuture();
-    await querySubscription.cancel();
+    Object? error;
+    StackTrace? stackTrace;
+    try {
+      await querySubscription.asFuture();
+    } catch (e, s) {
+      error = e;
+      stackTrace = s;
+      await querySubscription._done.future;
+    } finally {
+      await querySubscription.cancel();
+    }
+
+    if (error != null) {
+      Error.throwWithStackTrace(error, stackTrace!);
+    }
   }
 
   @override

pubspec.yaml

@@ -1,6 +1,6 @@
 name: postgres
 description: PostgreSQL database driver. Supports binary protocol, connection pooling and statement reuse.
-version: 3.5.11
+version: 3.5.12
 homepage: https://github.com/isoos/postgresql-dart
 topics:
   - sql

test/transaction_test.dart

@@ -836,5 +836,51 @@ void main() {
         }
       },
     );
+
+    test('failed BEGIN during runTx clears active transaction state', () async {
+      await conn.execute('BEGIN');
+      try {
+        await conn.execute('SELECT 1 FROM _nonexistent_xyz_ LIMIT 1');
+      } catch (_) {
+        // Leave the manually-started transaction in PostgreSQL's failed state.
+      }
+
+      await expectLater(
+        conn.runTx((tx) async {
+          await tx.execute('SELECT 1');
+        }),
+        throwsA(isA<PgException>()),
+      );
+
+      final rows = await conn.execute('SELECT 1');
+      expect(rows, [
+        [1],
+      ]);
+    });
+
+    test(
+      'ROLLBACK TO SAVEPOINT clears _transactionException so subsequent work commits',
+      () async {
+        await conn.runTx((tx) async {
+          await tx.execute('SAVEPOINT sp1');
+
+          try {
+            // 42P01 – relation does not exist
+            await tx.execute('SELECT 1 FROM _nonexistent_xyz_ LIMIT 1');
+          } catch (_) {
+            await tx.execute('ROLLBACK TO SAVEPOINT sp1');
+            await tx.execute('RELEASE SAVEPOINT sp1');
+          }
+
+          // This insert must commit; before the fix it was silently rolled back.
+          await tx.execute('INSERT INTO t (id) VALUES (42)');
+        });
+
+        final rows = await conn.execute('SELECT id FROM t WHERE id = 42');
+        expect(rows, [
+          [42],
+        ], reason: 'Insert after savepoint recovery should have been committed');
+      },
+    );
   });
 }