diff --git a/lib/core/database/mysql_connection.dart b/lib/core/database/mysql_connection.dart index 2927b94e..c066a19f 100644 --- a/lib/core/database/mysql_connection.dart +++ b/lib/core/database/mysql_connection.dart @@ -146,6 +146,27 @@ class MysqlConnection { MySQLConnection? _conn; Future? _connecting; + + /// Statements on this session run one at a time, here. The client would + /// otherwise poll for the previous statement and give up after 10 s, which + /// closed the session for every caller. + Future _statementQueue = Future.value(); + + /// Test seam: replaces the driver call of [execute]. + @visibleForTesting + Future Function( + String sql, + Map? params, + bool iterable, + Duration? timeout, + )? runStatementForTest; + + Future _serialized(Future Function() run) { + final previous = _statementQueue; + final turn = Completer(); + _statementQueue = turn.future; + return previous.then((_) => run()).whenComplete(() => turn.complete()); + } bool _isConnected = false; bool _inTransaction = false; @@ -441,18 +462,36 @@ class MysqlConnection { String sql, [ Map? params, bool iterable = false, - ]) async { - if (!isConnected || _conn == null) { - throw StateError('Not connected to MySQL'); - } - try { - final rs = await _conn!.execute(sql, params, iterable); - _noteTransactionSql(sql); - return rs; - } on TimeoutException { - unawaited(forceClose()); - rethrow; - } + ]) => + _run(sql, params, iterable, null); + + Future _run( + String sql, + Map? params, + bool iterable, + Duration? timeout, + ) { + return _serialized(() async { + final seam = runStatementForTest; + final c = _conn; + if (seam == null && (!isConnected || c == null)) { + throw StateError('Not connected to MySQL'); + } + try { + final pending = seam != null + ? seam(sql, params, iterable, timeout) + : c!.execute(sql, params, iterable); + final rs = timeout == null ? await pending : await pending.timeout(timeout); + _noteTransactionSql(sql); + return rs; + } on TimeoutException { + // The client cannot cancel a running statement, so the session is out of + // step after a timeout and is closed. Statements queued behind this one + // wait in the queue and are not affected. + unawaited(forceClose()); + rethrow; + } + }); } /// Whether this session has an open `START TRANSACTION` / `BEGIN`. @@ -498,22 +537,15 @@ class MysqlConnection { } } - /// Runs [execute] with an application-level [timeout] (driver limits still apply). + /// [execute] with [timeout] as the statement's own limit; it starts when the + /// statement starts, not while the statement waits behind another one. Future executeWithTimeout( String sql, { Duration? timeout, Map? params, bool iterable = false, - }) async { - final f = execute(sql, params, iterable); - if (timeout == null) return f; - try { - return await f.timeout(timeout); - } on TimeoutException { - unawaited(forceClose()); - rethrow; - } - } + }) => + _run(sql, params, iterable, timeout); /// Lists user-visible databases (excludes typical system schemas). Future> listDatabases() async { diff --git a/lib/core/database/postgres_connection.dart b/lib/core/database/postgres_connection.dart index 68bd4a6a..bc896a04 100644 --- a/lib/core/database/postgres_connection.dart +++ b/lib/core/database/postgres_connection.dart @@ -144,6 +144,22 @@ class PostgresConnection { Connection? _conn; Future? _connecting; + + /// Statements on this session run one at a time, here and not in the driver. + /// The driver starts a statement's timeout before the statement gets its turn, + /// so a statement that waited behind another one could cancel the other one. + Future _statementQueue = Future.value(); + + /// Test seam: replaces the driver call of [execute]. + @visibleForTesting + Future Function(String sql, Duration? timeout)? runStatementForTest; + + Future _serialized(Future Function() run) { + final previous = _statementQueue; + final turn = Completer(); + _statementQueue = turn.future; + return previous.then((_) => run()).whenComplete(() => turn.complete()); + } bool _isConnected = false; bool _inTransaction = false; @@ -391,37 +407,41 @@ class PostgresConnection { } } - /// Runs SQL on the underlying session. [timeout] overrides - /// [ConnectionSettings.queryTimeout] for this statement (see `postgres` - /// package). - Future execute(String sql, {Duration? timeout}) async { - if (!isConnected || _conn == null) { - throw StateError('Not connected to PostgreSQL'); - } - try { - final result = await _conn!.execute(sql, timeout: timeout); - _inTransaction = applyPostgresTransactionSql(_inTransaction, sql); - return result; - } on TimeoutException { - unawaited(forceClose()); - rethrow; - } + /// Runs SQL on the underlying session, one statement at a time. [timeout] + /// overrides [ConnectionSettings.queryTimeout] for this statement and starts + /// when the statement does. + /// + /// A server-side cancel (SQLSTATE 57014) arrives as a [PgException], and the + /// session stays usable. Only a bare [TimeoutException] closes it: then the + /// protocol may be out of step. + Future execute(String sql, {Duration? timeout}) { + return _serialized(() async { + final seam = runStatementForTest; + final c = _conn; + if (seam == null && (!isConnected || c == null)) { + throw StateError('Not connected to PostgreSQL'); + } + try { + final result = seam != null + ? await seam(sql, timeout) + : await c!.execute(sql, timeout: timeout); + _inTransaction = applyPostgresTransactionSql(_inTransaction, sql); + return result; + } on TimeoutException catch (e) { + if (e is! PgException) unawaited(forceClose()); + rethrow; + } + }); } - /// Runs [execute] with an application-level [timeout] (in addition to driver timeout). + /// [execute] with [timeout] as the statement's own limit. The limit starts + /// when the statement starts: time spent waiting behind another statement on + /// this session does not count against it. Future executeWithTimeout( String sql, { Duration? timeout, - }) async { - final f = execute(sql, timeout: timeout); - if (timeout == null) return f; - try { - return await f.timeout(timeout); - } on TimeoutException { - unawaited(forceClose()); - rethrow; - } - } + }) => + execute(sql, timeout: timeout); /// Whether the session has an open transaction. /// diff --git a/test/core/database/connection_timeout_protocol_test.dart b/test/core/database/connection_timeout_protocol_test.dart index 6a0a5e79..5f1a4c43 100644 --- a/test/core/database/connection_timeout_protocol_test.dart +++ b/test/core/database/connection_timeout_protocol_test.dart @@ -4,7 +4,6 @@ import 'package:mysql_client/mysql_client.dart'; import 'package:querya_desktop/core/database/mysql_connection.dart'; import 'package:querya_desktop/core/database/postgres_connection.dart'; import 'package:querya_desktop/core/database/sqlite_connection.dart'; -import 'package:postgres/postgres.dart' as pg; class FakeSlowMysqlConnection extends MysqlConnection { FakeSlowMysqlConnection({super.id = 1}) @@ -53,12 +52,6 @@ class FakeSlowPostgresConnection extends PostgresConnection { @override bool get isConnected => _connected; - @override - Future execute(String sql, {Duration? timeout}) async { - await Future.delayed(const Duration(seconds: 10)); - throw Exception('should not reach here'); - } - @override Future forceClose() async { forceCloseCount++; @@ -100,6 +93,9 @@ void main() { test('MysqlConnection.executeWithTimeout force-closes on timeout', () async { final conn = FakeSlowMysqlConnection(); expect(conn.isConnected, isTrue); + // The statement never answers: the timeout is the only way out. + conn.runStatementForTest = (sql, params, iterable, timeout) => + Completer().future; try { await conn.executeWithTimeout( @@ -116,9 +112,12 @@ void main() { expect(conn.isConnected, isFalse); }); - test('PostgresConnection.executeWithTimeout force-closes on TimeoutException', () async { + test('PostgresConnection.executeWithTimeout force-closes on a bare TimeoutException', () async { final conn = FakeSlowPostgresConnection(); expect(conn.isConnected, isTrue); + // The driver's own timer fires as a bare TimeoutException. + conn.runStatementForTest = (sql, timeout) => + Future.error(TimeoutException('driver timer', timeout)); try { await conn.executeWithTimeout( diff --git a/test/core/database/statement_queue_test.dart b/test/core/database/statement_queue_test.dart new file mode 100644 index 00000000..5f4c59f9 --- /dev/null +++ b/test/core/database/statement_queue_test.dart @@ -0,0 +1,135 @@ +import 'dart:async'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:mysql_client/mysql_client.dart'; +import 'package:postgres/postgres.dart' show PgException; +import 'package:querya_desktop/core/database/mysql_connection.dart'; +import 'package:querya_desktop/core/database/postgres_connection.dart'; + +/// Records force-closes instead of touching a socket. +class _ProbePostgres extends PostgresConnection { + _ProbePostgres() + : super( + id: 1, + name: 'probe', + host: 'localhost', + port: 5432, + database: 'postgres', + ); + + int forced = 0; + + @override + Future forceClose() async { + forced++; + } +} + +class _ProbeMysql extends MysqlConnection { + _ProbeMysql() + : super( + id: 1, + name: 'probe', + host: 'localhost', + port: 3306, + database: 'testdb', + ); + + int forced = 0; + + @override + Future forceClose() async { + forced++; + } +} + +void main() { + group('PostgreSQL statements on one session (#1216)', () { + test('a server-side cancel is a PgException and keeps the session', + () async { + final conn = _ProbePostgres(); + conn.runStatementForTest = (sql, timeout) => + Future.error(PgException('canceling statement due to user request')); + + await expectLater( + conn.execute('SELECT 1'), + throwsA(isA()), + ); + await Future.delayed(Duration.zero); + expect(conn.forced, 0); + }); + + test('a bare TimeoutException closes the session', () async { + final conn = _ProbePostgres(); + conn.runStatementForTest = (sql, timeout) => + Future.error(TimeoutException('driver timer', timeout)); + + await expectLater( + conn.execute('SELECT 1', timeout: const Duration(seconds: 1)), + throwsA(isA()), + ); + await Future.delayed(Duration.zero); + expect(conn.forced, 1); + }); + + test('a statement waits for the one before it on the same session', + () async { + final conn = _ProbePostgres(); + final log = []; + final gate = Completer(); + conn.runStatementForTest = (sql, timeout) async { + log.add('start $sql'); + if (sql == 'A') await gate.future; + log.add('end $sql'); + throw StateError('no result in this probe'); + }; + + final a = conn.execute('A').then((_) {}, onError: (_) {}); + final b = conn.execute('B').then((_) {}, onError: (_) {}); + await Future.delayed(Duration.zero); + expect(log, ['start A'], reason: 'B must not start while A runs'); + + gate.complete(); + await Future.wait([a, b]); + expect(log, ['start A', 'end A', 'start B', 'end B']); + }); + }); + + group('MySQL statements on one session (#1216)', () { + test('a statement waits for the one before it on the same session', + () async { + final conn = _ProbeMysql(); + final log = []; + final gate = Completer(); + conn.runStatementForTest = (sql, params, iterable, timeout) async { + log.add('start $sql'); + if (sql == 'A') await gate.future; + log.add('end $sql'); + throw StateError('no result in this probe'); + }; + + final a = conn.execute('A').then((_) {}, onError: (_) {}); + final b = conn.execute('B').then((_) {}, onError: (_) {}); + await Future.delayed(Duration.zero); + expect(log, ['start A']); + + gate.complete(); + await Future.wait([a, b]); + expect(log, ['start A', 'end A', 'start B', 'end B']); + }); + + test('a timed-out statement closes the session it ran on', () async { + final conn = _ProbeMysql(); + conn.runStatementForTest = (sql, params, iterable, timeout) => + Completer().future; + + await expectLater( + conn.executeWithTimeout('SELECT SLEEP(100)', + timeout: const Duration(milliseconds: 20)), + throwsA(isA()), + ); + await Future.delayed(Duration.zero); + expect(conn.forced, 1); + }); + }); +}