Throw exception if connection-level query executation is called while…

… inside a transaction callback. (#378)
isoos · Sep 13, 2024 · 6691c3d · 6691c3d
1 parent 80b9078
commit 6691c3d
Show file tree

Hide file tree

Showing 5 changed files with 29 additions and 72 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -13,6 +13,7 @@
     (for values where type is not specified).
   - `DatabaseInfo` tracks information about relations and oids (currently limited to `RelationMessage` caching).
 - **Behaviour / soft-breaking changes**:
+  - Preparing/executing a stamement on the main connection while in a `runTx` callback will throw an exception.
   - Deprecated `TupleDataColumn.data`, use `.value` instead (for binary protocol messages).
   - Deprecated some logical replication message parsing method.
   - Removed `@internal`-annotated methods from the public API of `ServerException` and `Severity`.

diff --git a/lib/src/v3/connection.dart b/lib/src/v3/connection.dart
@@ -105,6 +105,17 @@ abstract class _PgSessionBase implements Session {
   @override
   Future<void> get closed => _sessionClosedCompleter.future;
 
+  void _verifyStateBeforeQuery() {
+    if (_connection._isClosing || _sessionClosed) {
+      throw PgException(
+          'Attempting to execute query, but connection is not open.');
+    }
+    if (this == _connection && _connection._activeTransaction != null) {
+      throw PgException(
+          'Attempting to execute query on connection while inside a `runTx` call.');
+    }
+  }
+
   @override
   Future<Result> execute(
     Object query, {
@@ -113,10 +124,7 @@ abstract class _PgSessionBase implements Session {
     QueryMode? queryMode,
     Duration? timeout,
   }) async {
-    if (_connection._isClosing || _sessionClosed) {
-      throw PgException(
-          'Attempting to execute query, but connection is not open.');
-    }
+    _verifyStateBeforeQuery();
     final description = InternalQueryDescription.wrap(
       query,
       typeRegistry: _connection._settings.typeRegistry,
@@ -174,6 +182,7 @@ abstract class _PgSessionBase implements Session {
     Object query, {
     Duration? timeout,
   }) async {
+    _verifyStateBeforeQuery();
     return await _prepare(query, timeout: timeout);
   }
 

diff --git a/test/timeout_test.dart b/test/timeout_test.dart
@@ -43,16 +43,16 @@ void main() {
       expect(await conn.execute('SELECT * from t'), hasLength(0));
     });
 
-    test(
-        'Query on parent context for transaction completes (with error) after timeout',
+    test('Timeout is ignored when new statement is run on parent context',
         () async {
       try {
         await conn.runTx((ctx) async {
           await conn.execute('SELECT 1', timeout: Duration(seconds: 1));
           await ctx.execute('INSERT INTO t (id) VALUES (1)');
         });
         fail('unreachable');
-      } on TimeoutException {
+      } on PgException catch (e) {
+        expect(e.message, contains('runTx'));
         // ignore
       }
 

diff --git a/test/transaction_test.dart b/test/transaction_test.dart
@@ -50,38 +50,19 @@ void main() {
       ]);
     });
 
-    test('Query during transaction must wait until transaction is finished',
-        () async {
-      final orderEnsurer = [];
-      final nextCompleter = Completer.sync();
-      final outResult = conn.runTx((c) async {
-        orderEnsurer.add(1);
-        await c.execute('INSERT INTO t (id) VALUES (1)');
-        orderEnsurer.add(2);
-        nextCompleter.complete();
-        final result = await c.execute('SELECT id FROM t');
-        orderEnsurer.add(3);
-
-        return result;
-      });
-
-      await nextCompleter.future;
-      orderEnsurer.add(11);
-      await conn.execute('INSERT INTO t (id) VALUES (2)');
-      orderEnsurer.add(12);
-      final laterResults = await conn.execute('SELECT id FROM t');
-      orderEnsurer.add(13);
-
-      final firstResult = await outResult;
+    test('Connection query during transaction will throw exception.', () async {
+      try {
+        await conn.runTx((ctx) async {
+          await conn.execute('SELECT 1');
+          await ctx.execute('INSERT INTO t (id) VALUES (1)');
+        });
+        fail('unreachable');
+      } on PgException catch (e) {
+        expect(e.message, contains('runTx'));
+        // ignore
+      }
 
-      expect(orderEnsurer, [1, 2, 11, 3, 12, 13]);
-      expect(firstResult, [
-        [1]
-      ]);
-      expect(laterResults, [
-        [1],
-        [2]
-      ]);
+      expect(await conn.execute('SELECT * from t'), hasLength(0));
     });
 
     test('Make sure two simultaneous transactions cannot be interwoven',

diff --git a/test/v3_test.dart b/test/v3_test.dart
@@ -249,40 +249,6 @@ void main() {
         ]);
       });
 
-      test('Query during transaction must wait until transaction is finished',
-          () async {
-        final orderEnsurer = [];
-        final nextCompleter = Completer.sync();
-        final outResult = connection.runTx((c) async {
-          orderEnsurer.add(1);
-          await c.execute('INSERT INTO t (id) VALUES (1)');
-          orderEnsurer.add(2);
-          nextCompleter.complete();
-          final result = await c.execute('SELECT id FROM t');
-          orderEnsurer.add(3);
-
-          return result;
-        });
-
-        await nextCompleter.future;
-        orderEnsurer.add(11);
-        await connection.execute('INSERT INTO t (id) VALUES (2)');
-        orderEnsurer.add(12);
-        final laterResults = await connection.execute('SELECT id FROM t');
-        orderEnsurer.add(13);
-
-        final firstResult = await outResult;
-
-        expect(orderEnsurer, [1, 2, 11, 3, 12, 13]);
-        expect(firstResult, [
-          [1]
-        ]);
-        expect(laterResults, [
-          [1],
-          [2]
-        ]);
-      });
-
       test('Make sure two simultaneous transactions cannot be interwoven',
           () async {
         final orderEnsurer = [];