diff --git a/lib/core/database/database_error_mapper.dart b/lib/core/database/database_error_mapper.dart index a05f6570..a140e7a3 100644 --- a/lib/core/database/database_error_mapper.dart +++ b/lib/core/database/database_error_mapper.dart @@ -6,6 +6,7 @@ import 'package:querya_desktop/core/database/postgres_connection.dart'; import 'package:querya_desktop/core/database/querya_database_exception.dart'; import 'package:querya_desktop/core/database/redis_connection.dart'; import 'package:querya_desktop/core/database/sqlite_connection.dart'; +import 'package:querya_desktop/core/database/statement_queue.dart'; /// Which engine produced an error; only used to tailor the wording. enum DatabaseDriver { postgres, mysql, sqlite, mongodb, redis, extension } @@ -27,7 +28,15 @@ QueryaDatabaseException mapDatabaseError( QueryaDatabaseException? mapped; - if (_hasAny(lower, const [ + if (error is StatementWaitTimeoutException) { + mapped = ConnectionTimeoutException( + error.message, + remediationHint: 'Wait for the running statement to finish, or cancel ' + 'it, then try again', + originalError: error, + stackTrace: stackTrace, + ); + } else if (_hasAny(lower, const [ 'lock wait timeout', 'lock timeout', 'database is locked', diff --git a/lib/core/database/mysql_connection.dart b/lib/core/database/mysql_connection.dart index 7c8aaeb4..062d6656 100644 --- a/lib/core/database/mysql_connection.dart +++ b/lib/core/database/mysql_connection.dart @@ -2,6 +2,7 @@ import 'dart:async'; import 'package:flutter/foundation.dart'; import 'package:mysql_client/mysql_client.dart'; +import 'package:querya_desktop/core/database/statement_queue.dart'; import 'package:querya_desktop/core/database/database_error_mapper.dart'; import 'package:querya_desktop/core/database/mysql_result_cells.dart'; import 'package:querya_desktop/core/database/table_schema_meta.dart'; @@ -147,10 +148,11 @@ 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(); + /// Statements on this session run one at a time, here and not in the + /// driver: a statement's timeout starts when it starts, and waiting for the + /// one before it has its own limit (#1216). + @visibleForTesting + final statementQueue = StatementQueue(); /// Test seam: replaces the driver call of [execute]. @visibleForTesting @@ -161,12 +163,6 @@ class MysqlConnection { 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; @@ -485,7 +481,7 @@ class MysqlConnection { bool iterable, Duration? timeout, ) { - return _serialized(() async { + return statementQueue.run(() async { final seam = runStatementForTest; final c = _conn; if (seam == null && (!isConnected || c == null)) { diff --git a/lib/core/database/postgres_connection.dart b/lib/core/database/postgres_connection.dart index fcf4311f..077d2c50 100644 --- a/lib/core/database/postgres_connection.dart +++ b/lib/core/database/postgres_connection.dart @@ -3,6 +3,7 @@ import 'dart:io' show SecurityContext; import 'package:flutter/foundation.dart'; import 'package:postgres/postgres.dart'; +import 'package:querya_desktop/core/database/statement_queue.dart'; import 'package:querya_desktop/core/database/database_error_mapper.dart'; import 'package:querya_desktop/core/security/ssh_tunnel_config.dart'; import 'package:querya_desktop/core/security/ssh_tunnel_manager.dart'; @@ -145,21 +146,16 @@ 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(); + /// Statements on this session run one at a time, here and not in the + /// driver: a statement's timeout starts when it starts, and waiting for the + /// one before it has its own limit (#1216). + @visibleForTesting + final statementQueue = StatementQueue(); /// 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; @@ -408,7 +404,7 @@ class PostgresConnection { try { await connect(); if (_conn != null) { - await _conn!.execute('SELECT 1'); + await _queued((c) => c.execute('SELECT 1')); return (ok: true, error: null); } return (ok: false, error: 'Connection could not be established.'); @@ -429,7 +425,7 @@ class PostgresConnection { /// 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 { + return statementQueue.run(() async { final seam = runStatementForTest; final c = _conn; if (seam == null && (!isConnected || c == null)) { @@ -448,6 +444,19 @@ class PostgresConnection { }); } + /// A driver call on this session in its turn of [statementQueue], for the + /// metadata and stats queries. Without the queue the driver would start the + /// call's default timeout while it waits, and could cancel the statement + /// that is running (#1216). + Future _queued(Future Function(Connection c) call) => + statementQueue.run(() { + final c = _conn; + if (!isConnected || c == null) { + throw StateError('Not connected to PostgreSQL'); + } + return call(c); + }); + /// [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. @@ -467,7 +476,7 @@ class PostgresConnection { if (!isConnected || _conn == null) return null; if (_inTransaction) return true; try { - final r = await _conn!.execute(kPostgresOpenTransactionProbeSql); + final r = await _queued((c) => c.execute(kPostgresOpenTransactionProbeSql)); if (r.isEmpty) return false; final open = r.first[0] == true; _inTransaction = open; @@ -482,9 +491,9 @@ class PostgresConnection { if (!isConnected || _conn == null) { throw StateError('Not connected to PostgreSQL'); } - final result = await _conn!.execute( + final result = await _queued((c) => c.execute( "SELECT datname FROM pg_database WHERE datistemplate = false ORDER BY datname", - ); + )); return result.map((row) => row[0] as String).toList(); } @@ -492,11 +501,11 @@ class PostgresConnection { if (!isConnected || _conn == null) { throw StateError('Not connected to PostgreSQL'); } - final result = await _conn!.execute( + final result = await _queued((c) => c.execute( "SELECT schema_name FROM information_schema.schemata " "WHERE schema_name NOT IN ('pg_catalog', 'information_schema', 'pg_toast') " "ORDER BY schema_name", - ); + )); return result.map((row) => row[0] as String).toList(); } @@ -506,7 +515,7 @@ class PostgresConnection { if (!isConnected || _conn == null) { throw StateError('Not connected to PostgreSQL'); } - final result = await _conn!.execute( + final result = await _queued((c) => c.execute( "SELECT n.nspname::text, c.relname::text, " "CASE c.relkind " "WHEN 'r' THEN 'table' " @@ -521,7 +530,7 @@ class PostgresConnection { "AND n.nspname NOT LIKE 'pg_temp%' " "AND n.nspname NOT LIKE 'pg_toast_temp%' " "ORDER BY 1, 2", - ); + )); return [ for (final row in result) ( @@ -536,14 +545,14 @@ class PostgresConnection { if (!isConnected || _conn == null) { throw StateError('Not connected to PostgreSQL'); } - final result = await _conn!.execute( + final result = await _queued((c) => c.execute( Sql.named( "SELECT table_name FROM information_schema.tables " "WHERE table_schema = @schema AND table_type = 'BASE TABLE' " "ORDER BY table_name", ), parameters: {'schema': schema}, - ); + )); return result.map((row) => row[0] as String).toList(); } @@ -555,7 +564,7 @@ class PostgresConnection { if (!isConnected || _conn == null) { throw StateError('Not connected to PostgreSQL'); } - final colsRs = await _conn!.execute( + final colsRs = await _queued((c) => c.execute( Sql.named( 'SELECT column_name, data_type, udt_name, is_nullable, column_default, ' 'is_generated, is_identity, identity_generation ' @@ -564,9 +573,9 @@ class PostgresConnection { 'ORDER BY ordinal_position', ), parameters: {'schema': schema, 'table': table}, - ); + )); - final pkRs = await _conn!.execute( + final pkRs = await _queued((c) => c.execute( Sql.named( 'SELECT kcu.column_name ' 'FROM information_schema.table_constraints tc ' @@ -579,7 +588,7 @@ class PostgresConnection { 'ORDER BY kcu.ordinal_position', ), parameters: {'schema': schema, 'table': table}, - ); + )); final primaryKeys = pkRs.map((r) => r[0] as String).toList(); final columns = []; @@ -634,7 +643,7 @@ class PostgresConnection { if (!isConnected || _conn == null) { throw StateError('Not connected to PostgreSQL'); } - final result = await _conn!.execute( + final result = await _queued((c) => c.execute( Sql.named( 'SELECT c.reltuples FROM pg_catalog.pg_class c ' 'JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace ' @@ -643,7 +652,7 @@ class PostgresConnection { 'LIMIT 1', ), parameters: {'schema': schema, 'table': table}, - ); + )); if (result.isEmpty) return null; return postgresReltuplesEstimate(result.first[0]); } @@ -661,14 +670,14 @@ class PostgresConnection { if (!isConnected || _conn == null) { throw StateError('Not connected to PostgreSQL'); } - final result = await _conn!.execute( + final result = await _queued((c) => c.execute( Sql.named( "SELECT table_name FROM information_schema.views " "WHERE table_schema = @schema " "ORDER BY table_name", ), parameters: {'schema': schema}, - ); + )); return result.map((row) => row[0] as String).toList(); } @@ -676,14 +685,14 @@ class PostgresConnection { if (!isConnected || _conn == null) { throw StateError('Not connected to PostgreSQL'); } - final result = await _conn!.execute( + final result = await _queued((c) => c.execute( Sql.named( "SELECT routine_name FROM information_schema.routines " "WHERE routine_schema = @schema AND routine_type = 'FUNCTION' " "ORDER BY routine_name", ), parameters: {'schema': schema}, - ); + )); return result.map((row) => row[0] as String).toList(); } @@ -691,14 +700,14 @@ class PostgresConnection { if (!isConnected || _conn == null) { throw StateError('Not connected to PostgreSQL'); } - final result = await _conn!.execute( + final result = await _queued((c) => c.execute( Sql.named( "SELECT sequence_name FROM information_schema.sequences " "WHERE sequence_schema = @schema " "ORDER BY sequence_name", ), parameters: {'schema': schema}, - ); + )); return result.map((row) => row[0] as String).toList(); } @@ -706,7 +715,7 @@ class PostgresConnection { if (!isConnected || _conn == null) { throw StateError('Not connected to PostgreSQL'); } - final result = await _conn!.execute('SELECT version()'); + final result = await _queued((c) => c.execute('SELECT version()')); return result.first[0] as String; } @@ -717,40 +726,40 @@ class PostgresConnection { } final stats = {}; - final ver = await _conn!.execute('SELECT version()'); + final ver = await _queued((c) => c.execute('SELECT version()')); stats['version'] = ver.first[0] as String; - final settings = await _conn!.execute( + final settings = await _queued((c) => c.execute( "SELECT name, setting FROM pg_settings " "WHERE name IN ('max_connections','shared_buffers','work_mem'," "'effective_cache_size','server_version','data_directory'," "'listen_addresses','port','server_encoding','timezone')", - ); + )); final settingsMap = {}; for (final row in settings) { settingsMap[row[0] as String] = row[1] as String; } stats['settings'] = settingsMap; - final activity = await _conn!.execute( + final activity = await _queued((c) => c.execute( "SELECT count(*) AS total, " "count(*) FILTER (WHERE state = 'active') AS active, " "count(*) FILTER (WHERE state = 'idle') AS idle " "FROM pg_stat_activity", - ); + )); if (activity.isNotEmpty) { stats['connections_total'] = activity.first[0]; stats['connections_active'] = activity.first[1]; stats['connections_idle'] = activity.first[2]; } - final dbStats = await _conn!.execute( + final dbStats = await _queued((c) => c.execute( "SELECT datname, pg_database_size(datname) AS size, " "numbackends, xact_commit, xact_rollback, blks_read, blks_hit, " "tup_returned, tup_fetched, tup_inserted, tup_updated, tup_deleted " "FROM pg_stat_database WHERE datname NOT LIKE 'template%' " "ORDER BY datname", - ); + )); final dbList = >[]; for (final row in dbStats) { dbList.add({ @@ -771,18 +780,18 @@ class PostgresConnection { stats['databases'] = dbList; try { - final uptime = await _conn!.execute( + final uptime = await _queued((c) => c.execute( "SELECT extract(epoch from (now() - pg_postmaster_start_time()))::bigint", - ); + )); stats['uptime_seconds'] = uptime.first[0]; } catch (e) { debugPrint('PostgresConnection.getServerStats uptime: $e'); } try { - final dbSize = await _conn!.execute( + final dbSize = await _queued((c) => c.execute( "SELECT pg_database_size(current_database())", - ); + )); stats['current_db_size'] = dbSize.first[0]; } catch (e) { debugPrint('PostgresConnection.getServerStats database size: $e'); @@ -821,7 +830,7 @@ class PostgresConnection { if (!isConnected || _conn == null) { throw StateError('Not connected to PostgreSQL'); } - final result = await _conn!.execute( + final result = await _queued((c) => c.execute( Sql.named( r''' SELECT p.oid::regprocedure::text AS signature, @@ -833,7 +842,7 @@ ORDER BY p.oid ''', ), parameters: {'schema': schema, 'name': name}, - ); + )); return result .map( (row) => PgFunctionOverload( @@ -852,7 +861,7 @@ ORDER BY p.oid if (!isConnected || _conn == null) { throw StateError('Not connected to PostgreSQL'); } - final result = await _conn!.execute( + final result = await _queued((c) => c.execute( Sql.named( r''' SELECT last_value::text, start_value::text, min_value::text, max_value::text, @@ -863,7 +872,7 @@ WHERE schemaname = @schema AND sequencename = @name ''', ), parameters: {'schema': schema, 'name': name}, - ); + )); if (result.isEmpty) return null; final row = result.first; final sc = row[7] as String; @@ -897,13 +906,13 @@ WHERE schemaname = @schema AND sequencename = @name if (!isConnected || _conn == null) { throw StateError('Not connected to PostgreSQL'); } - final result = await _conn!.execute( + final result = await _queued((c) => c.execute( Sql.named( 'SELECT matviewname FROM pg_matviews WHERE schemaname = @schema ' 'ORDER BY matviewname', ), parameters: {'schema': schema}, - ); + )); return result.map((row) => row[0] as String).toList(); } @@ -912,14 +921,14 @@ WHERE schemaname = @schema AND sequencename = @name throw StateError('Not connected to PostgreSQL'); } final q = '${_quoteIdent(schema)}.${_quoteIdent(name)}'; - await _conn!.execute('REFRESH MATERIALIZED VIEW $q'); + await _queued((c) => c.execute('REFRESH MATERIALIZED VIEW $q')); } Future> listIndexesInSchema(String schema) async { if (!isConnected || _conn == null) { throw StateError('Not connected to PostgreSQL'); } - final result = await _conn!.execute( + final result = await _queued((c) => c.execute( Sql.named( r''' SELECT c.relname::text, @@ -936,7 +945,7 @@ ORDER BY c.relname, i.relname ''', ), parameters: {'schema': schema}, - ); + )); return result .map( (row) => PgIndexRow( @@ -960,7 +969,7 @@ ORDER BY c.relname, i.relname if (!isConnected || _conn == null) { throw StateError('Not connected to PostgreSQL'); } - final result = await _conn!.execute( + final result = await _queued((c) => c.execute( Sql.named( r''' SELECT c.relname::text, @@ -975,7 +984,7 @@ ORDER BY c.relname, t.tgname ''', ), parameters: {'schema': schema}, - ); + )); return result .map( (row) => PgTriggerRow( @@ -991,7 +1000,7 @@ ORDER BY c.relname, t.tgname if (!isConnected || _conn == null) { throw StateError('Not connected to PostgreSQL'); } - final result = await _conn!.execute( + final result = await _queued((c) => c.execute( Sql.named( r''' SELECT t.typname::text, @@ -1013,7 +1022,7 @@ ORDER BY t.typname ''', ), parameters: {'schema': schema}, - ); + )); return result .map( (row) => PgTypeRow( @@ -1028,9 +1037,9 @@ ORDER BY t.typname if (!isConnected || _conn == null) { throw StateError('Not connected to PostgreSQL'); } - final result = await _conn!.execute( + final result = await _queued((c) => c.execute( 'SELECT extname::text, extversion::text FROM pg_extension ORDER BY extname', - ); + )); return result .map( (row) => PgExtensionRow( @@ -1045,13 +1054,13 @@ ORDER BY t.typname if (!isConnected || _conn == null) { throw StateError('Not connected to PostgreSQL'); } - final result = await _conn!.execute( + final result = await _queued((c) => c.execute( r''' SELECT fdwname::text, fdwhandler::regproc::text FROM pg_foreign_data_wrapper ORDER BY fdwname ''', - ); + )); return result .map( (row) => PgFdwRow( @@ -1066,14 +1075,14 @@ ORDER BY fdwname if (!isConnected || _conn == null) { throw StateError('Not connected to PostgreSQL'); } - final result = await _conn!.execute( + final result = await _queued((c) => c.execute( r''' SELECT s.srvname::text, w.fdwname::text FROM pg_foreign_server s JOIN pg_foreign_data_wrapper w ON w.oid = s.srvfdw ORDER BY s.srvname ''', - ); + )); return result .map( (row) => PgForeignServerRow( @@ -1091,7 +1100,7 @@ ORDER BY s.srvname if (!isConnected || _conn == null) { throw StateError('Not connected to PostgreSQL'); } - final result = await _conn!.execute( + final result = await _queued((c) => c.execute( Sql.named( r''' SELECT grantee::text, privilege_type::text, is_grantable::text @@ -1101,7 +1110,7 @@ ORDER BY grantee, privilege_type ''', ), parameters: {'schema': schema, 'table': table}, - ); + )); return result .map( (row) => PgTablePrivilegeRow( diff --git a/lib/core/database/statement_queue.dart b/lib/core/database/statement_queue.dart new file mode 100644 index 00000000..fb4886e1 --- /dev/null +++ b/lib/core/database/statement_queue.dart @@ -0,0 +1,89 @@ +import 'dart:async'; + +/// Statements on one database session, run one at a time in our own layer. +/// +/// The drivers do not do this well: `postgres` starts a statement's timeout +/// before the statement gets the connection, so a statement waiting in line +/// could cancel the one running; `mysql_client` polls for the previous +/// statement and gives up after 10 s. Here a statement's own timeout starts +/// when it starts, and waiting has its own limit and its own error (#1216). +class StatementQueue { + StatementQueue({this.waitLimit = defaultWaitLimit}); + + /// How long a statement may wait for the ones before it. + static const defaultWaitLimit = Duration(minutes: 2); + + /// A statement still waiting after this fails with + /// [StatementWaitTimeoutException] and never runs; the statement it waited + /// for is not touched. Zero waits without a limit. + Duration waitLimit; + + Future _tail = Future.value(); + + /// Runs [statement] after every statement queued before it. + Future run(Future Function() statement) { + final previous = _tail; + final turn = Completer(); + _tail = turn.future; + + final result = Completer(); + final limit = waitLimit; + final waitTimer = limit > Duration.zero + ? Timer(limit, () { + if (!result.isCompleted) { + result.completeError(StatementWaitTimeoutException(limit)); + } + }) + : null; + + unawaited(previous.then((_) async { + waitTimer?.cancel(); + // Gave up waiting: skip the statement, pass the turn on. + if (result.isCompleted) { + turn.complete(); + return; + } + try { + result.complete(await statement()); + } catch (e, st) { + result.completeError(e, st); + } finally { + turn.complete(); + } + })); + return result.future; + } +} + +/// A statement waited longer than [limit] for another statement on the same +/// session and did not run. The session is fine: nothing was cancelled. +class StatementWaitTimeoutException implements Exception { + const StatementWaitTimeoutException(this.limit); + + final Duration limit; + + String get message => 'Waited ${formatStatementDuration(limit)} for the ' + 'connection: another statement is still running on it.'; + + @override + String toString() => message; +} + +/// What the user reads when a statement timed out: running too long or +/// waiting too long for the session. Null when [error] is neither. +String? statementTimeoutText(Object error) { + if (error is StatementWaitTimeoutException) return error.message; + if (error is TimeoutException) { + final d = error.duration; + // No duration: a server-side cancel the driver reports as a timeout + // (PostgreSQL 57014); its message says why. + return d == null + ? 'Query timed out: ${error.message ?? error}' + : 'Query timed out after ${formatStatementDuration(d)}'; + } + return null; +} + +/// `30 s`, or `250 ms` below a whole second. +String formatStatementDuration(Duration d) => + d.inMilliseconds % 1000 == 0 ? '${d.inSeconds} s' : '${d.inMilliseconds} ms'; diff --git a/lib/features/workspace/generic_sql_workspace.dart b/lib/features/workspace/generic_sql_workspace.dart index 77283938..151b1f93 100644 --- a/lib/features/workspace/generic_sql_workspace.dart +++ b/lib/features/workspace/generic_sql_workspace.dart @@ -1,3 +1,4 @@ +import 'package:querya_desktop/core/database/statement_queue.dart'; import 'package:querya_desktop/core/database/sql_statement_splitter.dart'; import 'dart:async'; import 'dart:io'; @@ -446,7 +447,7 @@ class GenericSqlWorkspaceState extends material.State { if (mounted) { invalidatePane(session); setState(() { - session.error = 'Query timed out: ${e.message ?? e}'; + session.error = statementTimeoutText(e); session.running = false; }); } @@ -798,7 +799,7 @@ class GenericSqlWorkspaceState extends material.State { if (mounted) { invalidatePane(session); setState(() { - session.error = 'Query timed out: ${e.message ?? e}'; + session.error = statementTimeoutText(e); session.running = false; }); QueryaShellStatus.instance.endBusy(); @@ -860,9 +861,8 @@ class GenericSqlWorkspaceState extends material.State { return at == null ? const [] : [at]; } - static String _runErrorText(Object e) => e is TimeoutException - ? 'Query timed out: ${e.message ?? e}' - : e.toString(); + static String _runErrorText(Object e) => + statementTimeoutText(e) ?? e.toString(); /// Shows the query plan of the selection (or the whole editor text) in the /// result grid, one plan line per row. diff --git a/test/core/database/statement_queue_test.dart b/test/core/database/statement_queue_test.dart index 5f4c59f9..f422189a 100644 --- a/test/core/database/statement_queue_test.dart +++ b/test/core/database/statement_queue_test.dart @@ -5,6 +5,7 @@ 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'; +import 'package:querya_desktop/core/database/statement_queue.dart'; /// Records force-closes instead of touching a socket. class _ProbePostgres extends PostgresConnection { @@ -132,4 +133,181 @@ void main() { expect(conn.forced, 1); }); }); + + group('PostgreSQL acceptance (#1216)', () { + test('after a server-side cancel the next statement runs on the session', + () async { + final conn = _ProbePostgres(); + final ran = []; + conn.runStatementForTest = (sql, timeout) { + ran.add(sql); + if (sql == 'A') { + return Future.error( + PgException('canceling statement due to user request')); + } + return Future.error(StateError('no result in this probe')); + }; + + await expectLater(conn.execute('A'), throwsA(isA())); + await expectLater(conn.execute('B'), throwsA(isA())); + expect(ran, ['A', 'B']); + expect(conn.forced, 0); + }); + + test( + 'a statement longer than the next one\'s timeout is not cancelled; ' + 'the next timeout starts with the next statement', () async { + final conn = _ProbePostgres(); + final log = []; + final gate = Completer(); + conn.runStatementForTest = (sql, timeout) async { + log.add('start $sql timeout=${timeout?.inMilliseconds}'); + if (sql == 'A') await gate.future; + log.add('end $sql'); + throw StateError('done $sql'); + }; + + Object? aError; + final a = conn.execute('A').then((_) {}, onError: (Object e) { + aError = e; + }); + final b = conn + .execute('B', timeout: const Duration(milliseconds: 20)) + .then((_) {}, onError: (_) {}); + // A runs well past B's 20 ms. + await Future.delayed(const Duration(milliseconds: 80)); + expect(log, ['start A timeout=null']); + + gate.complete(); + await Future.wait([a, b]); + // A finished on its own: B's timeout did not cancel it, and the driver got + // B's 20 ms only when B started. + expect((aError as StateError).message, 'done A'); + expect(log, [ + 'start A timeout=null', + 'end A', + 'start B timeout=20', + 'end B', + ]); + expect(conn.forced, 0); + }); + + test('a bare TimeoutException closes only the session it ran on', + () async { + final broken = _ProbePostgres(); + final other = _ProbePostgres(); + broken.runStatementForTest = (sql, timeout) => + Future.error(TimeoutException('driver timer', timeout)); + other.runStatementForTest = + (sql, timeout) => Future.error(StateError('no result in this probe')); + + await expectLater( + broken.execute('SELECT 1', timeout: const Duration(seconds: 1)), + throwsA(isA()), + ); + await expectLater(other.execute('SELECT 1'), throwsA(isA())); + await Future.delayed(Duration.zero); + expect(broken.forced, 1); + expect(other.forced, 0); + }); + }); + + group('MySQL acceptance (#1216)', () { + testWidgets('a query waiting more than 10 s behind another keeps the session', + (tester) 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: (_) {}); + // The client used to give up after 10 s of polling and close the session. + await tester.pump(const Duration(seconds: 11)); + expect(log, ['start A']); + + gate.complete(); + await tester.pump(); + await Future.wait([a, b]); + expect(log, ['start A', 'end A', 'start B', 'end B']); + expect(conn.forced, 0); + }); + }); + + group('StatementQueue', () { + testWidgets( + 'a statement that waits past the limit fails on its own and does not ' + 'run; the running one and the ones after it are untouched', + (tester) async { + final queue = StatementQueue(waitLimit: const Duration(minutes: 2)); + final log = []; + final gate = Completer(); + + final a = queue.run(() async { + log.add('A'); + await gate.future; + return 'a'; + }); + Object? bError; + final b = queue.run(() async { + log.add('B'); + return 'b'; + }).then((_) {}, onError: (Object e) { + bError = e; + }); + + await tester.pump(const Duration(minutes: 2, seconds: 1)); + await b; + expect(bError, isA()); + expect('$bError', contains('Waited 120 s for the connection')); + + final c = queue.run(() async { + log.add('C'); + return 'c'; + }); + gate.complete(); + await tester.pump(); + expect(await a, 'a'); + expect(await c, 'c'); + expect(log, ['A', 'C'], reason: 'B gave up and never ran'); + }); + + test('zero waits without a limit', () async { + final queue = StatementQueue(waitLimit: Duration.zero); + final gate = Completer(); + final a = queue.run(() => gate.future.then((_) => 1)); + final b = queue.run(() async => 2); + await Future.delayed(const Duration(milliseconds: 20)); + gate.complete(); + expect(await a, 1); + expect(await b, 2); + }); + }); + + group('timeout texts', () { + test('a running timeout and a wait timeout read differently', () { + expect( + statementTimeoutText( + TimeoutException('driver', const Duration(seconds: 30))), + 'Query timed out after 30 s', + ); + expect( + statementTimeoutText( + const StatementWaitTimeoutException(Duration(seconds: 120))), + startsWith('Waited 120 s for the connection'), + ); + expect( + statementTimeoutText(TimeoutException('canceling statement')), + 'Query timed out: canceling statement', + ); + expect(statementTimeoutText(StateError('x')), isNull); + expect(formatStatementDuration(const Duration(milliseconds: 250)), + '250 ms'); + }); + }); }