Only reset "is timed out" flag in public API methods.

This fixes a bug where a timed-out query would throw QueryInterrupted instead of the expected CommandTimeoutExpired because MySqlCommand.ExecuteReader would return a valid MySqlDataReader object, and the first call to MySqlDataReader.Read would clear the IsTimedOut flag (before a subsequent call to Read processed the pending error packet).

Author: bgrainger

src/MySqlConnector/MySqlCommand.cs

@@ -260,6 +260,7 @@ public override Task<int> ExecuteNonQueryAsync(CancellationToken cancellationTok
 
 		internal async Task<int> ExecuteNonQueryAsync(IOBehavior ioBehavior, CancellationToken cancellationToken)
 		{
+			Volatile.Write(ref m_commandTimedOut, false);
 			this.ResetCommandTimeout();
 			using var registration = ((ICancellableCommand) this).RegisterCancel(cancellationToken);
 			using var reader = await ExecuteReaderNoResetTimeoutAsync(CommandBehavior.Default, ioBehavior, cancellationToken).ConfigureAwait(false);
@@ -277,6 +278,7 @@ internal async Task<int> ExecuteNonQueryAsync(IOBehavior ioBehavior, Cancellatio
 
 		internal async Task<object?> ExecuteScalarAsync(IOBehavior ioBehavior, CancellationToken cancellationToken)
 		{
+			Volatile.Write(ref m_commandTimedOut, false);
 			this.ResetCommandTimeout();
 			using var registration = ((ICancellableCommand) this).RegisterCancel(cancellationToken);
 			var hasSetResult = false;
@@ -306,6 +308,7 @@ protected override async Task<DbDataReader> ExecuteDbDataReaderAsync(CommandBeha
 
 		internal async Task<MySqlDataReader> ExecuteReaderAsync(CommandBehavior behavior, IOBehavior ioBehavior, CancellationToken cancellationToken)
 		{
+			Volatile.Write(ref m_commandTimedOut, false);
 			this.ResetCommandTimeout();
 			using var registration = ((ICancellableCommand) this).RegisterCancel(cancellationToken);
 			return await ExecuteReaderNoResetTimeoutAsync(behavior, ioBehavior, cancellationToken).ConfigureAwait(false);
@@ -364,8 +367,6 @@ public override ValueTask DisposeAsync()
 
 		void ICancellableCommand.SetTimeout(int milliseconds)
 		{
-			Volatile.Write(ref m_commandTimedOut, false);
-
 			if (m_cancelTimerId != 0)
 				TimerQueue.Instance.Remove(m_cancelTimerId);
 

tests/MySqlConnector.Tests/CancellationTests.cs

@@ -53,7 +53,7 @@ public void Execute(int step, int method)
 
 			[SkipCITheory]
 			[MemberData(nameof(GetAsyncMethodSteps))]
-			public async Task ExecuteAsyncs(int step, int method)
+			public async Task ExecuteAsync(int step, int method)
 			{
 				using var connection = new MySqlConnection(m_csb.ConnectionString);
 				connection.Open();
@@ -74,6 +74,53 @@ public async Task ExecuteAsyncs(int step, int method)
 			}
 		}
 
+		public class CancelBufferedWithCommandTimeout : CancellationTests
+		{
+			[SkipCITheory]
+			[MemberData(nameof(GetSyncMethodSteps))]
+			public void Execute(int step, int method)
+			{
+				using var connection = new MySqlConnection(m_csb.ConnectionString);
+				connection.Open();
+				using var command = connection.CreateCommand();
+				command.CommandTimeout = 1;
+				command.CommandText = $"SELECT 0, 4000, {step}, 2;";
+				var stopwatch = Stopwatch.StartNew();
+				var ex = Assert.Throws<MySqlException>(() => s_executeMethods[method](command));
+				Assert.InRange(stopwatch.ElapsedMilliseconds, 900, 1500);
+				Assert.Equal(MySqlErrorCode.CommandTimeoutExpired, ex.ErrorCode);
+				var inner = Assert.IsType<MySqlException>(ex.InnerException);
+				Assert.Equal(MySqlErrorCode.QueryInterrupted, inner.ErrorCode);
+
+				// connection should still be usable
+				Assert.Equal(ConnectionState.Open, connection.State);
+				command.CommandText = "SELECT 1;";
+				Assert.Equal(1, command.ExecuteScalar());
+			}
+
+			[SkipCITheory]
+			[MemberData(nameof(GetAsyncMethodSteps))]
+			public async Task ExecuteAsync(int step, int method)
+			{
+				using var connection = new MySqlConnection(m_csb.ConnectionString);
+				connection.Open();
+				using var command = connection.CreateCommand();
+				command.CommandTimeout = 1;
+				command.CommandText = $"SELECT 0, 4000, {step}, 2;";
+				var stopwatch = Stopwatch.StartNew();
+				var ex = await Assert.ThrowsAsync<MySqlException>(async () => await s_executeAsyncMethods[method](command, default));
+				Assert.InRange(stopwatch.ElapsedMilliseconds, 900, 1500);
+				Assert.Equal(MySqlErrorCode.CommandTimeoutExpired, ex.ErrorCode);
+				var inner = Assert.IsType<MySqlException>(ex.InnerException);
+				Assert.Equal(MySqlErrorCode.QueryInterrupted, inner.ErrorCode);
+
+				// connection should still be usable
+				Assert.Equal(ConnectionState.Open, connection.State);
+				command.CommandText = "SELECT 1;";
+				Assert.Equal(1, command.ExecuteScalar());
+			}
+		}
+
 		public class CancelWithCancel : CancellationTests
 		{
 			[SkipCITheory]

tests/MySqlConnector.Tests/FakeMySqlServerConnection.cs

@@ -97,6 +97,7 @@ public async Task RunAsync(TcpClient client, CancellationToken token)
 								var pauseStep = int.Parse(match.Groups[3].Value);
 								var flags = int.Parse(match.Groups[4].Value);
 								var ignoreCancellation = (flags & 1) == 1;
+								var bufferOutput = (flags & 2) == 2;
 
 								var data = new byte[number.Length + 1];
 								data[0] = (byte) number.Length;
@@ -120,18 +121,34 @@ public async Task RunAsync(TcpClient client, CancellationToken token)
 									new byte[] { 0xFE, 0, 0, 2, 0 }, // EOF
 								};
 
-								var queryInterrupted = false;
-								for (var step = 1; step < packets.Length && !queryInterrupted; step++)
+								if (bufferOutput)
 								{
-									if (pauseStep == step || pauseStep == -1)
+									// if 'bufferOutput' is set, perform the delay immediately then send all the output afterwards, as though it were buffered on the server
+									var queryInterrupted = false;
+									if (ignoreCancellation)
+										await Task.Delay(delay, token);
+									else
+										queryInterrupted = CancelQueryEvent.Wait(delay, token);
+
+									for (var step = 1; step < pauseStep; step++)
+										await SendAsync(stream, step, x => x.Write(packets[step]));
+									await SendAsync(stream, pauseStep, x => x.Write(packets[queryInterrupted ? 0 : pauseStep]));
+								}
+								else
+								{
+									var queryInterrupted = false;
+									for (var step = 1; step < packets.Length && !queryInterrupted; step++)
 									{
-										if (ignoreCancellation)
-											await Task.Delay(delay, token);
-										else
-											queryInterrupted = CancelQueryEvent.Wait(delay, token);
-									}
+										if (pauseStep == step || pauseStep == -1)
+										{
+											if (ignoreCancellation)
+												await Task.Delay(delay, token);
+											else
+												queryInterrupted = CancelQueryEvent.Wait(delay, token);
+										}
 
-									await SendAsync(stream, step, x => x.Write(packets[queryInterrupted ? 0 : step]));
+										await SendAsync(stream, step, x => x.Write(packets[queryInterrupted ? 0 : step]));
+									}
 								}
 							}
 							else if ((match = Regex.Match(query, @"^KILL QUERY ([0-9]+)(;|$)", RegexOptions.IgnoreCase)).Success)