ADO stabilization: normalize provider-specific cancellation to OCE, add best-effort MySQL timeout cancel, wrap/track readers and auto-drain before Commit to fix multi-result Tx across SQL Server/Npgsql/MySQL, plus EF Core Npgsql column mapping & minor analyzer tweaks

Author: KeelMatrix

.editorconfig

@@ -45,6 +45,7 @@ dotnet_code_quality.CA1826.exclude_ordefault_methods = true
 [tests/**.cs]
 dotnet_diagnostic.CA1707.severity = none   # disable “no underscores in identifiers” for test names
 dotnet_diagnostic.CS1591.severity = none   # disable “missing XML documentation” for test code
+dotnet_diagnostic.IDE0290.severity = none  # disable “Use primary constructor”
 
 # sample apps under /samples
 [samples/**.cs]
@@ -60,6 +61,9 @@ dotnet_diagnostic.CS1591.severity = none   # disable “missing XML documentatio
 [src/KeelMatrix.QueryWatch/**.cs]
 dotnet_diagnostic.CA1510.severity = none   # disable “prefer ArgumentNullException.ThrowIfNull” requirement
 dotnet_diagnostic.CA1513.severity = none   # disable “prefer nameof(...) in exceptions” requirement
+dotnet_diagnostic.CA2249.severity = none   # disable “Use 'string.Contains' instead of 'string.IndexOf' to improve readability”
+dotnet_diagnostic.IDE0290.severity = none  # disable “Use primary constructor”
+dotnet_diagnostic.RCS1001.severity = none  # disable “Add braces to if statement”
 
 # Project: Contracts — src/KeelMatrix.QueryWatch.Contracts
 [src/KeelMatrix.QueryWatch.Contracts/**.cs]

src/KeelMatrix.QueryWatch/Ado/QueryWatchCommand.cs

@@ -1,3 +1,5 @@
+// Copyright (c) KeelMatrix
+
 #nullable enable
 using System.Data;
 using System.Data.Common;
@@ -6,14 +8,16 @@
 
 namespace KeelMatrix.QueryWatch.Ado {
     /// <summary>
-    /// Delegating <see cref="System.Data.Common.DbCommand"/> that measures execution and records into a session.
+    /// Delegating <see cref="DbCommand"/> that measures execution and records into a session.
     /// </summary>
     public sealed class QueryWatchCommand : DbCommand {
         private readonly DbCommand _inner;
         private readonly QueryWatchSession _session;
 
         // Keep track of the wrapper connection (so getters surface the wrapper instance).
         private QueryWatchConnection? _wrapperConnection;
+        // Keep track of a wrapper transaction so we can track readers for draining before Commit().
+        private QueryWatchTransaction? _wrapperTransaction;
 
         /// <summary>
         /// Initializes a new wrapper over an inner <see cref="DbCommand"/>.
@@ -28,6 +32,14 @@ public QueryWatchCommand(DbCommand inner, QueryWatchSession session, DbConnectio
             _wrapperConnection = wrapperConnection as QueryWatchConnection;
         }
 
+        private sealed class CompositeDisposer : IDisposable {
+            private readonly CancellationTokenSource _cts;
+            private readonly IDisposable _reg;
+            public CompositeDisposer(CancellationTokenSource cts, IDisposable reg) { _cts = cts; _reg = reg; }
+            public bool IsCancelled => _cts.IsCancellationRequested;
+            public void Dispose() { _reg.Dispose(); _cts.Dispose(); }
+        }
+
         #region Property delegation
 
         /// <inheritdoc />
@@ -69,13 +81,20 @@ protected override DbConnection? DbConnection {
 
         /// <inheritdoc />
         protected override DbTransaction? DbTransaction {
-            get => _inner.Transaction;
+            get {
+                // Prefer the wrapper if we have it and it still matches the inner transaction.
+                if (_wrapperTransaction is not null && ReferenceEquals(_inner.Transaction, _wrapperTransaction.Inner))
+                    return _wrapperTransaction;
+                return _inner.Transaction;
+            }
             set {
                 if (value is QueryWatchTransaction wrapped) {
                     _inner.Transaction = wrapped.Inner;
+                    _wrapperTransaction = wrapped;
                 }
                 else {
                     _inner.Transaction = value;
+                    _wrapperTransaction = null;
                 }
             }
         }
@@ -133,54 +152,135 @@ private void RecordFailure(TimeSpan elapsed, Exception ex) {
             _session.Record(text, elapsed, meta);
         }
 
+        private static Exception NormalizeCancellation(Exception ex, CancellationToken token) {
+            if (!token.IsCancellationRequested) return ex;
+
+            // SQL Server
+            var full = ex.GetType()?.FullName ?? string.Empty;
+            if (full.IndexOf("SqlClient.SqlException", StringComparison.OrdinalIgnoreCase) >= 0)
+                return new OperationCanceledException("Command was cancelled.", ex, token);
+
+            // Npgsql: PostgresException with SqlState 57014 (query_canceled)
+            var typeName = ex.GetType().FullName ?? "";
+            if ((typeName ?? string.Empty).IndexOf("Npgsql.PostgresException", StringComparison.OrdinalIgnoreCase) >= 0) {
+                try {
+                    var sqlStateProp = ex.GetType().GetProperty("SqlState");
+                    var sqlState = sqlStateProp?.GetValue(ex) as string;
+                    if (string.Equals(sqlState, "57014", StringComparison.Ordinal))
+                        return new OperationCanceledException("Command was cancelled.", ex, token);
+                }
+                catch { /* ignore reflection issues */ }
+            }
+
+            // MySQL: MySqlConnector.MySqlException often thrown, or TaskCanceled/Timeout
+            if ((typeName ?? string.Empty).IndexOf("MySqlConnector.MySqlException", StringComparison.OrdinalIgnoreCase) >= 0 ||
+                ex is TaskCanceledException || ex is TimeoutException)
+                return new OperationCanceledException("Command was cancelled.", ex, token);
+
+            // Fallback: if token says cancelled, prefer OCE so callers can handle uniformly.
+            return new OperationCanceledException("Command was cancelled.", ex, token);
+        }
+
+        private static bool IsMySql(DbCommand cmd)
+            => cmd.GetType().Namespace?.StartsWith("MySqlConnector", StringComparison.Ordinal) == true;
+
+        /// <summary>Registers a callback that calls provider Cancel() when token is cancelled.</summary>
+        private static CancellationTokenRegistration RegisterCancelOnToken(DbCommand cmd, CancellationToken token) {
+            if (!token.CanBeCanceled)
+                return default;
+            try {
+                return token.Register(static state => {
+                    try { ((DbCommand)state!).Cancel(); } catch { /* best-effort */ }
+                }, cmd);
+            }
+            catch { return default; }
+        }
+
+        /// <summary>For providers that may not enforce CommandTimeout deterministically (MySQL),
+        /// schedule a Cancel() at timeout expiry (seconds). No effect for zero/infinite timeout.</summary>
+        private static CompositeDisposer? BeginTimeoutCancelIfNeeded(DbCommand cmd) {
+            try {
+                if (!IsMySql(cmd))
+                    return null;               // scope to MySQL only
+                int seconds = cmd.CommandTimeout;
+                if (seconds <= 0)
+                    return null;                // 0 == infinite
+                var cts = new CancellationTokenSource(TimeSpan.FromSeconds(seconds));
+                var reg = cts.Token.Register(static state => {
+                    try { ((DbCommand)state!).Cancel(); } catch { /* best-effort */ }
+                }, cmd);
+                return new CompositeDisposer(cts, reg);
+            }
+            catch { return null; }
+        }
+
         #endregion
 
         #region Execute overrides (sync)
 
         /// <inheritdoc />
         public override int ExecuteNonQuery() {
             var sw = Stopwatch.StartNew();
+            IDisposable? tmo = BeginTimeoutCancelIfNeeded(_inner);
             try {
-                return _inner.ExecuteNonQuery();
+                var result = _inner.ExecuteNonQuery();
+                if (tmo is CompositeDisposer s && s.IsCancelled)
+                    throw new TimeoutException("CommandTimeout elapsed and the command was cancelled.");
+                return result;
             }
             catch (Exception ex) {
                 sw.Stop();
                 RecordFailure(sw.Elapsed, ex);
                 throw;
             }
             finally {
+                tmo?.Dispose();
                 if (sw.IsRunning) { sw.Stop(); RecordSuccess(sw.Elapsed); }
             }
         }
 
         /// <inheritdoc />
         public override object? ExecuteScalar() {
             var sw = Stopwatch.StartNew();
+            IDisposable? tmo = BeginTimeoutCancelIfNeeded(_inner);
             try {
-                return _inner.ExecuteScalar();
+                var result = _inner.ExecuteScalar();
+                if (tmo is CompositeDisposer s && s.IsCancelled)
+                    throw new TimeoutException("CommandTimeout elapsed and the command was cancelled.");
+                return result;
             }
             catch (Exception ex) {
                 sw.Stop();
                 RecordFailure(sw.Elapsed, ex);
                 throw;
             }
             finally {
+                tmo?.Dispose();
                 if (sw.IsRunning) { sw.Stop(); RecordSuccess(sw.Elapsed); }
             }
         }
 
         /// <inheritdoc />
         protected override DbDataReader ExecuteDbDataReader(CommandBehavior behavior) {
             var sw = Stopwatch.StartNew();
+            IDisposable? tmo = BeginTimeoutCancelIfNeeded(_inner);
             try {
-                return _inner.ExecuteReader(behavior);
+                var r = _inner.ExecuteReader(behavior);
+                if (tmo is CompositeDisposer s && s.IsCancelled) {
+                    r.Dispose();
+                    throw new TimeoutException("CommandTimeout elapsed and the command was cancelled.");
+                }
+                var tx = _wrapperTransaction;
+                tx?.TrackReader(r);
+                return new QueryWatchDataReader(r, tx);
             }
             catch (Exception ex) {
                 sw.Stop();
                 RecordFailure(sw.Elapsed, ex);
                 throw;
             }
             finally {
+                tmo?.Dispose();
                 if (sw.IsRunning) { sw.Stop(); RecordSuccess(sw.Elapsed); }
             }
         }
@@ -190,68 +290,102 @@ protected override DbDataReader ExecuteDbDataReader(CommandBehavior behavior) {
         #region Execute overrides (async)
 
         /// <inheritdoc />
-        public override System.Threading.Tasks.Task<int> ExecuteNonQueryAsync(System.Threading.CancellationToken cancellationToken) {
+        public override Task<int> ExecuteNonQueryAsync(CancellationToken cancellationToken) {
             return ExecuteNonQueryAsyncCore(cancellationToken);
         }
 
-        private async System.Threading.Tasks.Task<int> ExecuteNonQueryAsyncCore(System.Threading.CancellationToken token) {
+        private async Task<int> ExecuteNonQueryAsyncCore(CancellationToken token) {
             var sw = Stopwatch.StartNew();
+            CancellationTokenRegistration reg = default;
             try {
-                return await _inner.ExecuteNonQueryAsync(token).ConfigureAwait(false);
+                reg = RegisterCancelOnToken(_inner, token);
+                var result = await _inner.ExecuteNonQueryAsync(token).ConfigureAwait(false);
+                if (IsMySql(_inner) && token.IsCancellationRequested)
+                    throw new OperationCanceledException("Command was cancelled.", token);
+                return result;
             }
             catch (Exception ex) {
                 sw.Stop();
                 RecordFailure(sw.Elapsed, ex);
-                throw;
+                throw NormalizeCancellation(ex, token);
             }
             finally {
+#if NET8_0_OR_GREATER
+                await reg.DisposeAsync();
+#else
+                reg.Dispose();
+#endif
                 if (sw.IsRunning) { sw.Stop(); RecordSuccess(sw.Elapsed); }
             }
         }
 
         /// <inheritdoc />
-        public override System.Threading.Tasks.Task<object?> ExecuteScalarAsync(System.Threading.CancellationToken cancellationToken) {
+        public override Task<object?> ExecuteScalarAsync(CancellationToken cancellationToken) {
             return ExecuteScalarAsyncCore(cancellationToken);
         }
 
-        private async System.Threading.Tasks.Task<object?> ExecuteScalarAsyncCore(System.Threading.CancellationToken token) {
+        private async Task<object?> ExecuteScalarAsyncCore(CancellationToken token) {
             var sw = Stopwatch.StartNew();
+            CancellationTokenRegistration reg = default;
             try {
-                return await _inner.ExecuteScalarAsync(token).ConfigureAwait(false);
+                reg = RegisterCancelOnToken(_inner, token);
+                var obj = await _inner.ExecuteScalarAsync(token).ConfigureAwait(false);
+                if (IsMySql(_inner) && token.IsCancellationRequested)
+                    throw new OperationCanceledException("Command was cancelled.", token);
+                return obj;
             }
             catch (Exception ex) {
                 sw.Stop();
                 RecordFailure(sw.Elapsed, ex);
-                throw;
+                throw NormalizeCancellation(ex, token);
             }
             finally {
+#if NET8_0_OR_GREATER
+                await reg.DisposeAsync();
+#else
+                reg.Dispose();
+#endif
                 if (sw.IsRunning) { sw.Stop(); RecordSuccess(sw.Elapsed); }
             }
         }
 
         /// <inheritdoc />
-        protected override System.Threading.Tasks.Task<DbDataReader> ExecuteDbDataReaderAsync(CommandBehavior behavior, System.Threading.CancellationToken cancellationToken) {
-            return ExecuteDbDataReaderAsyncCore(behavior, cancellationToken);
-        }
-
-        private async System.Threading.Tasks.Task<DbDataReader> ExecuteDbDataReaderAsyncCore(CommandBehavior behavior, System.Threading.CancellationToken token) {
+        protected override async Task<DbDataReader> ExecuteDbDataReaderAsync(CommandBehavior behavior, CancellationToken cancellationToken) {
             var sw = Stopwatch.StartNew();
+            CancellationTokenRegistration reg = default;
             try {
-                return await _inner.ExecuteReaderAsync(behavior, token).ConfigureAwait(false);
+                reg = RegisterCancelOnToken(_inner, cancellationToken);
+                var r = await _inner.ExecuteReaderAsync(behavior, cancellationToken).ConfigureAwait(false);
+                if (IsMySql(_inner) && cancellationToken.IsCancellationRequested) {
+#if NET8_0_OR_GREATER
+                    await r.DisposeAsync();
+#else
+                    r.Dispose();
+#endif
+                    throw new OperationCanceledException("Command was cancelled.", cancellationToken);
+                }
+                var tx = _wrapperTransaction;
+                tx?.TrackReader(r);
+                return new QueryWatchDataReader(r, tx);
             }
             catch (Exception ex) {
                 sw.Stop();
                 RecordFailure(sw.Elapsed, ex);
-                throw;
+                throw NormalizeCancellation(ex, cancellationToken);
             }
             finally {
+#if NET8_0_OR_GREATER
+                await reg.DisposeAsync();
+#else
+                reg.Dispose();
+#endif
                 if (sw.IsRunning) { sw.Stop(); RecordSuccess(sw.Elapsed); }
             }
         }
 
         #endregion
 
-        #region Boilerplate
+        #region Misc
 
         /// <inheritdoc />
         public override void Cancel() => _inner.Cancel();
@@ -262,12 +396,6 @@ private async System.Threading.Tasks.Task<DbDataReader> ExecuteDbDataReaderAsync
         /// <inheritdoc />
         protected override DbParameter CreateDbParameter() => _inner.CreateParameter();
 
-        /// <inheritdoc />
-        protected override void Dispose(bool disposing) {
-            if (disposing) _inner.Dispose();
-            base.Dispose(disposing);
-        }
-
         #endregion
     }
 }