Hand Lite's analysis token to the store reads it is meant to abandon (#2443)

Darling/Darling.Tests/DarlingAnalysisStoreTests.cs

@@ -222,12 +222,12 @@ public async Task PlanFetcher_DegradesToNull_OnResolverMiss_AndSkipsResolveOnEmp
            to null without touching any server (Lite's ServerManager-miss semantics). */
         var resolved = false;
         var fetcher = new PgPlanFetcher(_ => { resolved = true; return null; });
-        Assert.Null(await fetcher.FetchPlanXmlAsync(TestServerId, "0x0600FF00"));
+        Assert.Null(await fetcher.FetchPlanXmlAsync(TestServerId, "0x0600FF00", TestContext.Current.CancellationToken));
         Assert.True(resolved);
 
         /* An empty plan handle short-circuits before the resolver is consulted. */
         resolved = false;
-        Assert.Null(await fetcher.FetchPlanXmlAsync(TestServerId, ""));
+        Assert.Null(await fetcher.FetchPlanXmlAsync(TestServerId, "", TestContext.Current.CancellationToken));
         Assert.False(resolved);
     }
 

Darling/PerformanceMonitor.Darling.Analysis/PgBlockingPairRowQuery.cs

@@ -9,6 +9,7 @@
 using System;
 using System.Collections.Generic;
 using System.Data.Common;
+using System.Threading;
 using System.Threading.Tasks;
 using Npgsql;
 using PerformanceMonitor.Analysis;
@@ -125,8 +126,15 @@ ORDER BY event_time DESC
     /// fragments as the blocked-process-report queries, against v_dmv_blocking_snapshots, so <see cref="Read"/>
     /// maps it unchanged. Takes a command factory so the caller's connection runs it — the Lite shape, kept so
     /// the future drill-down/viewer slices call it identically.
+    ///
+    /// <para>#2443: the token is required, not defaulted. Three callers share this fetch and two of
+    /// them are on the analysis pass; a default would have let either keep passing nothing while the
+    /// signature claimed the read was abandonable. The viewer's call is the one that legitimately has
+    /// no pass to abandon, and it says so at its own call site rather than here.</para>
     /// </summary>
-    internal static async Task AppendDmvSnapshotRowsAsync(Func<NpgsqlCommand> createCommand, List<BlockingPairRow> rows, int serverId, DateTime start, DateTime end)
+    internal static async Task AppendDmvSnapshotRowsAsync(
+        Func<NpgsqlCommand> createCommand, List<BlockingPairRow> rows, int serverId, DateTime start, DateTime end,
+        CancellationToken cancellationToken)
     {
         var dmv = new List<BlockingPairRow>();
         using (var cmd = createCommand())
@@ -136,8 +144,8 @@ internal static async Task AppendDmvSnapshotRowsAsync(Func<NpgsqlCommand> create
             cmd.Parameters.AddWithValue(DateTime.SpecifyKind(start, DateTimeKind.Unspecified));
             cmd.Parameters.AddWithValue(DateTime.SpecifyKind(end, DateTimeKind.Unspecified));
 
-            using var reader = await cmd.ExecuteReaderAsync();
-            while (await reader.ReadAsync())
+            using var reader = await cmd.ExecuteReaderAsync(cancellationToken);
+            while (await reader.ReadAsync(cancellationToken))
                 dmv.Add(Read(reader));
         }
 

Darling/PerformanceMonitor.Darling.Analysis/PgDrillDownCollector.Blocking.cs

@@ -29,16 +29,16 @@ ORDER BY collection_time DESC
 
     private async Task CollectTopDeadlocks(AnalysisFinding finding, AnalysisContext context)
     {
-        await using var connection = await _postgres.OpenConnectionAsync();
+        await using var connection = await _postgres.OpenConnectionAsync(context.CancellationToken);
 
         using var cmd = new NpgsqlCommand(TopDeadlocksSql, connection);
         cmd.Parameters.AddWithValue(context.ServerId);
         cmd.Parameters.AddWithValue(AsNaive(context.TimeRangeStart));
         cmd.Parameters.AddWithValue(AsNaive(context.TimeRangeEnd));
 
         var items = new List<object>();
-        using var reader = await cmd.ExecuteReaderAsync();
-        while (await reader.ReadAsync())
+        using var reader = await cmd.ExecuteReaderAsync(context.CancellationToken);
+        while (await reader.ReadAsync(context.CancellationToken))
         {
             /* #1140: parse the involved objects from the graph for the dedup fingerprint + a readable
                Objects field. The raw graph XML is NOT surfaced (it would bloat the alert detail). */
@@ -88,16 +88,16 @@ ORDER BY wait_time_ms DESC
 
     private async Task CollectTopBlockingChains(AnalysisFinding finding, AnalysisContext context)
     {
-        await using var connection = await _postgres.OpenConnectionAsync();
+        await using var connection = await _postgres.OpenConnectionAsync(context.CancellationToken);
 
         using var cmd = new NpgsqlCommand(TopBlockingChainsSql, connection);
         cmd.Parameters.AddWithValue(context.ServerId);
         cmd.Parameters.AddWithValue(AsNaive(context.TimeRangeStart));
         cmd.Parameters.AddWithValue(AsNaive(context.TimeRangeEnd));
 
         var items = new List<object>();
-        using var reader = await cmd.ExecuteReaderAsync();
-        while (await reader.ReadAsync())
+        using var reader = await cmd.ExecuteReaderAsync(context.CancellationToken);
+        while (await reader.ReadAsync(context.CancellationToken))
         {
             items.Add(new
             {
@@ -141,24 +141,25 @@ ORDER BY event_time DESC
     /// </summary>
     private async Task CollectReconstructedBlockingChains(AnalysisFinding finding, AnalysisContext context)
     {
-        await using var connection = await _postgres.OpenConnectionAsync();
+        await using var connection = await _postgres.OpenConnectionAsync(context.CancellationToken);
 
         using var cmd = new NpgsqlCommand(ReconstructedChainsSql, connection);
         cmd.Parameters.AddWithValue(context.ServerId);
         cmd.Parameters.AddWithValue(AsNaive(context.TimeRangeStart));
         cmd.Parameters.AddWithValue(AsNaive(context.TimeRangeEnd));
 
         var rows = new List<BlockingPairRow>();
-        using (var reader = await cmd.ExecuteReaderAsync())
+        using (var reader = await cmd.ExecuteReaderAsync(context.CancellationToken))
         {
-            while (await reader.ReadAsync())
+            while (await reader.ReadAsync(context.CancellationToken))
                 rows.Add(PgBlockingPairRowQuery.Read(reader));
         }
 
         // Always-on DMV blocking snapshot fallback. Merge BEFORE the empty check so DMV-only blocking
         // (blocked-process-report unavailable, e.g. AWS RDS) still reconstructs.
         await PgBlockingPairRowQuery.AppendDmvSnapshotRowsAsync(
-            connection.CreateCommand, rows, context.ServerId, context.TimeRangeStart, context.TimeRangeEnd);
+            connection.CreateCommand, rows, context.ServerId, context.TimeRangeStart, context.TimeRangeEnd,
+            context.CancellationToken);
 
         if (rows.Count == 0) return;
 
@@ -207,16 +208,16 @@ ORDER BY total_wait_ms DESC
 
     private async Task CollectLockModeBreakdown(AnalysisFinding finding, AnalysisContext context)
     {
-        await using var connection = await _postgres.OpenConnectionAsync();
+        await using var connection = await _postgres.OpenConnectionAsync(context.CancellationToken);
 
         using var cmd = new NpgsqlCommand(LockModeBreakdownSql, connection);
         cmd.Parameters.AddWithValue(context.ServerId);
         cmd.Parameters.AddWithValue(AsNaive(context.TimeRangeStart));
         cmd.Parameters.AddWithValue(AsNaive(context.TimeRangeEnd));
 
         var items = new List<object>();
-        using var reader = await cmd.ExecuteReaderAsync();
-        while (await reader.ReadAsync())
+        using var reader = await cmd.ExecuteReaderAsync(context.CancellationToken);
+        while (await reader.ReadAsync(context.CancellationToken))
         {
             items.Add(new
             {

Darling/PerformanceMonitor.Darling.Analysis/PgDrillDownCollector.Config.cs

@@ -28,14 +28,14 @@ FROM v_database_config
 
     private async Task CollectConfigIssues(AnalysisFinding finding, AnalysisContext context)
     {
-        await using var connection = await _postgres.OpenConnectionAsync();
+        await using var connection = await _postgres.OpenConnectionAsync(context.CancellationToken);
 
         using var cmd = new NpgsqlCommand(ConfigIssuesSql, connection);
         cmd.Parameters.AddWithValue(context.ServerId);
 
         var items = new List<object>();
-        using var reader = await cmd.ExecuteReaderAsync();
-        while (await reader.ReadAsync())
+        using var reader = await cmd.ExecuteReaderAsync(context.CancellationToken);
+        while (await reader.ReadAsync(context.CancellationToken))
         {
             var issues = new List<string>();
             if (!reader.IsDBNull(2) && reader.GetBoolean(2)) issues.Add("auto_shrink ON");

Darling/PerformanceMonitor.Darling.Analysis/PgDrillDownCollector.Plans.cs

@@ -49,22 +49,27 @@ private async Task CollectPlanAnalysis(AnalysisFinding finding, AnalysisContext
         string? planHandle = null;
         try
         {
-            await using var connection = await _postgres.OpenConnectionAsync();
+            await using var connection = await _postgres.OpenConnectionAsync(context.CancellationToken);
 
             using var cmd = new NpgsqlCommand(PlanHandleLookupSql, connection);
             cmd.Parameters.AddWithValue(context.ServerId);
             cmd.Parameters.AddWithValue(queryHash);
 
-            using var reader = await cmd.ExecuteReaderAsync();
-            if (await reader.ReadAsync() && !reader.IsDBNull(0))
+            using var reader = await cmd.ExecuteReaderAsync(context.CancellationToken);
+            if (await reader.ReadAsync(context.CancellationToken) && !reader.IsDBNull(0))
                 planHandle = reader.GetString(0);
         }
-        catch { return; }
+        catch (Exception ex) when (!AnalysisShutdown.IsExpectedAbandon(ex, context.CancellationToken))
+        {
+            /* No plan_handle for this hash — the fetch below has nothing to ask for. An abandonment
+               is NOT swallowed here (#2443). */
+            return;
+        }
 
         if (string.IsNullOrEmpty(planHandle)) return;
 
         // Fetch plan XML live from SQL Server
-        var planXml = await _planFetcher.FetchPlanXmlAsync(context.ServerId, planHandle);
+        var planXml = await _planFetcher.FetchPlanXmlAsync(context.ServerId, planHandle, context.CancellationToken);
         if (string.IsNullOrEmpty(planXml)) return;
 
         try
@@ -142,15 +147,15 @@ private async Task CollectPlanAdvisoryDetail(AnalysisFinding finding, AnalysisCo
             /* PG port: Lite scopes the connection in a block to release its DuckDB read lock
                before the CPU-only parse; the scoping is kept so the connection closes before
                the parse, even though PG holds no lock. */
-            await using (var connection = await _postgres.OpenConnectionAsync())
+            await using (var connection = await _postgres.OpenConnectionAsync(context.CancellationToken))
             {
                 using var cmd = new NpgsqlCommand(PlanAdvisoryXmlSql, connection);
                 cmd.Parameters.AddWithValue(context.ServerId);
                 cmd.Parameters.AddWithValue(AsNaive(context.TimeRangeStart));
                 cmd.Parameters.AddWithValue(AsNaive(context.TimeRangeEnd));
 
-                using var reader = await cmd.ExecuteReaderAsync();
-                while (await reader.ReadAsync())
+                using var reader = await cmd.ExecuteReaderAsync(context.CancellationToken);
+                while (await reader.ReadAsync(context.CancellationToken))
                 {
                     if (!reader.IsDBNull(0))
                         planXmls.Add(reader.GetString(0));
@@ -190,9 +195,10 @@ private async Task CollectPlanAdvisoryDetail(AnalysisFinding finding, AnalysisCo
                     .ToList();
             }
         }
-        catch
+        catch (Exception ex) when (!AnalysisShutdown.IsExpectedAbandon(ex, context.CancellationToken))
         {
             // Plan read/parse can fail on malformed XML — skip, the detail is best-effort.
+            // An abandonment is NOT swallowed here (#2443).
         }
     }
 }

Darling/PerformanceMonitor.Darling.Analysis/PgDrillDownCollector.Queries.cs

@@ -48,7 +48,7 @@ ORDER BY cpu_time_ms DESC
     private async Task CollectQueriesAtSpike(AnalysisFinding finding, AnalysisContext context)
     {
         // Find the peak CPU time, then get queries active within 2 minutes of it
-        await using var connection = await _postgres.OpenConnectionAsync();
+        await using var connection = await _postgres.OpenConnectionAsync(context.CancellationToken);
 
         // Step 1: Find when the spike occurred
         using var peakCmd = new NpgsqlCommand(SpikePeakSql, connection);
@@ -59,9 +59,9 @@ private async Task CollectQueriesAtSpike(AnalysisFinding finding, AnalysisContex
 
         DateTime? peakTime = null;
         int peakCpu = 0;
-        using (var peakReader = await peakCmd.ExecuteReaderAsync())
+        using (var peakReader = await peakCmd.ExecuteReaderAsync(context.CancellationToken))
         {
-            if (await peakReader.ReadAsync())
+            if (await peakReader.ReadAsync(context.CancellationToken))
             {
                 peakTime = peakReader.GetDateTime(0);
                 peakCpu = peakReader.GetInt32(1);
@@ -78,9 +78,9 @@ private async Task CollectQueriesAtSpike(AnalysisFinding finding, AnalysisContex
         queryCmd.Parameters.AddWithValue(AsNaive(peakTime.Value.AddMinutes(2)));
 
         var items = new List<object>();
-        using (var reader = await queryCmd.ExecuteReaderAsync())
+        using (var reader = await queryCmd.ExecuteReaderAsync(context.CancellationToken))
         {
-            while (await reader.ReadAsync())
+            while (await reader.ReadAsync(context.CancellationToken))
             {
                 items.Add(new
                 {
@@ -126,7 +126,7 @@ ORDER BY total_cpu_us DESC
 
     private async Task CollectTopCpuQueries(AnalysisFinding finding, AnalysisContext context)
     {
-        await using var connection = await _postgres.OpenConnectionAsync();
+        await using var connection = await _postgres.OpenConnectionAsync(context.CancellationToken);
 
         using var cmd = new NpgsqlCommand(TopCpuQueriesSql, connection);
         cmd.CommandTimeout = DrillDownCommandTimeoutSeconds;
@@ -135,8 +135,8 @@ private async Task CollectTopCpuQueries(AnalysisFinding finding, AnalysisContext
         cmd.Parameters.AddWithValue(AsNaive(context.TimeRangeEnd));
 
         var items = new List<object>();
-        using var reader = await cmd.ExecuteReaderAsync();
-        while (await reader.ReadAsync())
+        using var reader = await cmd.ExecuteReaderAsync(context.CancellationToken);
+        while (await reader.ReadAsync(context.CancellationToken))
         {
             items.Add(new
             {
@@ -168,7 +168,7 @@ ORDER BY total_spills DESC
 
     private async Task CollectTopSpillingQueries(AnalysisFinding finding, AnalysisContext context)
     {
-        await using var connection = await _postgres.OpenConnectionAsync();
+        await using var connection = await _postgres.OpenConnectionAsync(context.CancellationToken);
 
         using var cmd = new NpgsqlCommand(TopSpillingQueriesSql, connection);
         cmd.CommandTimeout = DrillDownCommandTimeoutSeconds;
@@ -177,8 +177,8 @@ private async Task CollectTopSpillingQueries(AnalysisFinding finding, AnalysisCo
         cmd.Parameters.AddWithValue(AsNaive(context.TimeRangeEnd));
 
         var items = new List<object>();
-        using var reader = await cmd.ExecuteReaderAsync();
-        while (await reader.ReadAsync())
+        using var reader = await cmd.ExecuteReaderAsync(context.CancellationToken);
+        while (await reader.ReadAsync(context.CancellationToken))
         {
             items.Add(new
             {
@@ -248,7 +248,7 @@ ORDER BY worker_ratio DESC
     /// </summary>
     private async Task CollectParameterSensitiveQueries(AnalysisFinding finding, AnalysisContext context)
     {
-        await using var connection = await _postgres.OpenConnectionAsync();
+        await using var connection = await _postgres.OpenConnectionAsync(context.CancellationToken);
 
         using var cmd = new NpgsqlCommand(ParameterSensitiveSql, connection);
         cmd.CommandTimeout = DrillDownCommandTimeoutSeconds;
@@ -257,8 +257,8 @@ private async Task CollectParameterSensitiveQueries(AnalysisFinding finding, Ana
         cmd.Parameters.AddWithValue(AsNaive(context.TimeRangeEnd));
 
         var items = new List<object>();
-        using var reader = await cmd.ExecuteReaderAsync();
-        while (await reader.ReadAsync())
+        using var reader = await cmd.ExecuteReaderAsync(context.CancellationToken);
+        while (await reader.ReadAsync(context.CancellationToken))
         {
             items.Add(new
             {
@@ -492,7 +492,7 @@ ORDER BY regression_factor DESC
     /// </summary>
     private async Task CollectRegressedQueries(AnalysisFinding finding, AnalysisContext context)
     {
-        await using var connection = await _postgres.OpenConnectionAsync();
+        await using var connection = await _postgres.OpenConnectionAsync(context.CancellationToken);
 
         using var cmd = new NpgsqlCommand(RegressedQueriesSql, connection);
         cmd.CommandTimeout = DrillDownCommandTimeoutSeconds;
@@ -505,8 +505,8 @@ private async Task CollectRegressedQueries(AnalysisFinding finding, AnalysisCont
         cmd.Parameters.AddWithValue(AsNaive(context.TimeRangeEnd));
 
         var items = new List<object>();
-        using var reader = await cmd.ExecuteReaderAsync();
-        while (await reader.ReadAsync())
+        using var reader = await cmd.ExecuteReaderAsync(context.CancellationToken);
+        while (await reader.ReadAsync(context.CancellationToken))
         {
             items.Add(new
             {
@@ -567,7 +567,7 @@ private async Task CollectBadActorDetail(AnalysisFinding finding, AnalysisContex
         var queryHash = finding.RootFactKey.Replace("BAD_ACTOR_", "");
         if (string.IsNullOrEmpty(queryHash)) return;
 
-        await using var connection = await _postgres.OpenConnectionAsync();
+        await using var connection = await _postgres.OpenConnectionAsync(context.CancellationToken);
 
         using var cmd = new NpgsqlCommand(BadActorDetailSql, connection);
         cmd.CommandTimeout = DrillDownCommandTimeoutSeconds;
@@ -576,8 +576,8 @@ private async Task CollectBadActorDetail(AnalysisFinding finding, AnalysisContex
         cmd.Parameters.AddWithValue(AsNaive(context.TimeRangeEnd));
         cmd.Parameters.AddWithValue(queryHash);
 
-        using var reader = await cmd.ExecuteReaderAsync();
-        if (await reader.ReadAsync())
+        using var reader = await cmd.ExecuteReaderAsync(context.CancellationToken);
+        if (await reader.ReadAsync(context.CancellationToken))
         {
             finding.DrillDown!["bad_actor_query"] = new
             {
@@ -610,7 +610,7 @@ ORDER BY waiter_count DESC
 
     private async Task CollectPendingGrants(AnalysisFinding finding, AnalysisContext context)
     {
-        await using var connection = await _postgres.OpenConnectionAsync();
+        await using var connection = await _postgres.OpenConnectionAsync(context.CancellationToken);
 
         using var cmd = new NpgsqlCommand(PendingGrantsSql, connection);
         cmd.CommandTimeout = DrillDownCommandTimeoutSeconds;
@@ -619,8 +619,8 @@ private async Task CollectPendingGrants(AnalysisFinding finding, AnalysisContext
         cmd.Parameters.AddWithValue(AsNaive(context.TimeRangeEnd));
 
         var items = new List<object>();
-        using var reader = await cmd.ExecuteReaderAsync();
-        while (await reader.ReadAsync())
+        using var reader = await cmd.ExecuteReaderAsync(context.CancellationToken);
+        while (await reader.ReadAsync(context.CancellationToken))
         {
             items.Add(new
             {