Thread cancellation through 10 PostgreSQL web reads (#4203)

Darling/Darling.Tests/McpPayloadContractCensusTests.cs

@@ -1717,7 +1717,7 @@ public void TheSixMovedSites_FetchOnePastTheCap_AndObserveThroughBoundPage(strin
     public void TheAutovacuumRunHistory_BindsToItsCapAndObserves()
     {
         var source = StripComments(ToolSources().Single(s => s.File == "DarlingMcpPgAutovacuumTools.cs").Source);
-        Assert.Contains("RunsReadPerTable + 1);", source, StringComparison.Ordinal);
+        Assert.Contains("RunsReadPerTable + 1, cancellationToken);", source, StringComparison.Ordinal);
         Assert.Contains("McpHelpers.BoundPage(runs, RunsReadPerTable)", source, StringComparison.Ordinal);
         Assert.DoesNotContain("history_capped", source, StringComparison.Ordinal);
         Assert.DoesNotContain(">= RunsReadPerTable", source, StringComparison.Ordinal);

Darling/Darling.Tests/PgLogEventMetricsTests.cs

@@ -442,7 +442,7 @@ public void TheReaders_NameTheColumns_AndTheDescriptionsSayWhatANullMeans()
         Assert.Contains("log_autovacuum_min_duration", healthDescription, StringComparison.Ordinal);
         Assert.Contains("never zero", healthDescription, StringComparison.Ordinal);
         /* No new parameter: the run history rides the tool's existing window and page. */
-        Assert.Equal(new[] { "postgres", "server_name", "hours_back", "limit", "as_of" }, health.GetParameters().Select(p => p.Name));
+        Assert.Equal(new[] { "postgres", "server_name", "hours_back", "limit", "as_of" }, health.GetParameters().Where(p => p.ParameterType != typeof(CancellationToken)).Select(p => p.Name));
 
         var tool = RepoFile.ReadRepoFile("Darling", "PerformanceMonitor.Darling.Service", "Mcp", "DarlingMcpPgAutovacuumTools.cs");
         Assert.Contains("GetAutovacuumRunsAsync(", tool, StringComparison.Ordinal);

Darling/PerformanceMonitor.Darling.Service/DarlingWebEndpoints.cs

@@ -139,21 +139,11 @@ public static class DarlingWebEndpoints
         "get_analysis_findings",
         "get_file_io_trend",
         "get_perfmon_trend",
-        "get_pg_autovacuum_health",
         "get_pg_column_stats",
         "get_pg_cpu_utilization",
-        "get_pg_database_stats",
         "get_pg_index_bloat",
-        "get_pg_index_usage",
-        "get_pg_io_stats",
-        "get_pg_kernel_stats",
         "get_pg_log_events",
         "get_pg_logging_audit",
-        "get_pg_predicate_stats",
-        "get_pg_replication_stats",
-        "get_pg_session_states",
-        "get_pg_wraparound_risk",
-        "get_pg_xmin_horizon",
         "get_store_host",
     };
 
@@ -3105,18 +3095,18 @@ cannot parse exactly rather than silently matching nothing. */
             ["get_pg_plans"] = (c, pg, an) => DarlingMcpPgPlanTools.GetPgPlans(pg, Server(c), Hours(c, 24), Rows(c, "limit", 10), Str(c, "query_id"), AsOf(c), cancellationToken: c.RequestAborted),
             ["get_pg_plan_capture_readiness"] = (c, pg, an) => DarlingMcpPgPlanTools.GetPgPlanCaptureReadiness(pg, Server(c), Hours(c, 24), Rows(c, "limit", 25), as_of: AsOf(c), cancellationToken: c.RequestAborted),
             ["get_pg_logging_audit"] = (c, pg, an) => DarlingMcpPgLoggingAuditTools.GetPgLoggingAudit(pg, Server(c)),
-            ["get_pg_wraparound_risk"] = (c, pg, an) => DarlingMcpPgWraparoundTools.GetPgWraparoundRisk(pg, Server(c), Hours(c, 24), as_of: AsOf(c)),
-            ["get_pg_xmin_horizon"] = (c, pg, an) => DarlingMcpPgXminTools.GetPgXminHorizon(pg, Server(c), Hours(c, 24), as_of: AsOf(c)),
+            ["get_pg_wraparound_risk"] = (c, pg, an) => DarlingMcpPgWraparoundTools.GetPgWraparoundRisk(pg, Server(c), Hours(c, 24), as_of: AsOf(c), cancellationToken: c.RequestAborted),
+            ["get_pg_xmin_horizon"] = (c, pg, an) => DarlingMcpPgXminTools.GetPgXminHorizon(pg, Server(c), Hours(c, 24), as_of: AsOf(c), cancellationToken: c.RequestAborted),
             ["get_pg_replication_slots"] = (c, pg, an) => DarlingMcpPgSlotTools.GetPgReplicationSlots(pg, Server(c), Hours(c, 24), as_of: AsOf(c), cancellationToken: c.RequestAborted),
-            ["get_pg_autovacuum_health"] = (c, pg, an) => DarlingMcpPgAutovacuumTools.GetPgAutovacuumHealth(pg, Server(c), Hours(c, 24), Rows(c, "limit", 20), as_of: AsOf(c)),
-            ["get_pg_io_stats"] = (c, pg, an) => DarlingMcpPgIoTools.GetPgIoStats(pg, Server(c), Hours(c, 24), Rows(c, "limit", 20), as_of: AsOf(c)),
+            ["get_pg_autovacuum_health"] = (c, pg, an) => DarlingMcpPgAutovacuumTools.GetPgAutovacuumHealth(pg, Server(c), Hours(c, 24), Rows(c, "limit", 20), as_of: AsOf(c), cancellationToken: c.RequestAborted),
+            ["get_pg_io_stats"] = (c, pg, an) => DarlingMcpPgIoTools.GetPgIoStats(pg, Server(c), Hours(c, 24), Rows(c, "limit", 20), as_of: AsOf(c), cancellationToken: c.RequestAborted),
             ["get_pg_wait_stats"] = (c, pg, an) => DarlingMcpPgWaitTools.GetPgWaitStats(pg, Server(c), Hours(c, 24), Rows(c, "limit", 20), as_of: AsOf(c), cancellationToken: c.RequestAborted),
             ["get_pg_cpu_utilization"] = (c, pg, an) => OptionalInt(c, "bucket_minutes", out var bucketMinutes)
                 ? DarlingMcpPgCpuUtilizationTools.GetPgCpuUtilization(pg, Server(c), Hours(c, 4), AsOf(c), bucketMinutes, TrendBudget.Chart)
                 : UnparseableParam("bucket_minutes"),
             ["get_pg_wait_sampling"] = (c, pg, an) => DarlingMcpPgWaitSamplingTools.GetPgWaitSampling(pg, Server(c), Hours(c, 24), Rows(c, "limit", 20), as_of: AsOf(c), cancellationToken: c.RequestAborted),
-            ["get_pg_kernel_stats"] = (c, pg, an) => DarlingMcpPgKernelStatsTools.GetPgKernelStats(pg, Server(c), Hours(c, 24), Rows(c, "limit", 20), as_of: AsOf(c)),
-            ["get_pg_predicate_stats"] = (c, pg, an) => DarlingMcpPgPredicateTools.GetPgPredicateStats(pg, Server(c), Hours(c, 24), Rows(c, "limit", 25), as_of: AsOf(c)),
+            ["get_pg_kernel_stats"] = (c, pg, an) => DarlingMcpPgKernelStatsTools.GetPgKernelStats(pg, Server(c), Hours(c, 24), Rows(c, "limit", 20), as_of: AsOf(c), cancellationToken: c.RequestAborted),
+            ["get_pg_predicate_stats"] = (c, pg, an) => DarlingMcpPgPredicateTools.GetPgPredicateStats(pg, Server(c), Hours(c, 24), Rows(c, "limit", 25), as_of: AsOf(c), cancellationToken: c.RequestAborted),
             ["get_pg_index_bloat"] = (c, pg, an) => DarlingMcpPgIndexTools.GetPgIndexBloat(pg, Server(c), Hours(c, 168), Rows(c, "limit", 25), QueryBool(c, "answered_only", false), as_of: AsOf(c)),
             ["get_pg_column_stats"] = (c, pg, an) => DarlingMcpPgIndexTools.GetPgColumnStats(pg, Server(c), Hours(c, 168), Rows(c, "limit", 25), as_of: AsOf(c)),
             ["get_pg_buffer_usage"] = (c, pg, an) => DarlingMcpPgServerStateTools.GetPgBufferUsage(pg, Server(c), Hours(c, 24), Rows(c, "limit", 25), as_of: AsOf(c), cancellationToken: c.RequestAborted),
@@ -3138,12 +3128,12 @@ cannot parse exactly rather than silently matching nothing. */
             ["get_pg_database_trend"] = (c, pg, an) => OptionalInt(c, "bucket_minutes", out var bucketMinutes)
                 ? DarlingMcpPgTrendTools.GetPgDatabaseTrend(pg, Server(c), Str(c, "database"), Hours(c, 24), AsOf(c), bucketMinutes, TrendBudget.Chart, c.RequestAborted)
                 : UnparseableParam("bucket_minutes"),
-            ["get_pg_replication_stats"] = (c, pg, an) => DarlingMcpPgReplicationStatsTools.GetPgReplicationStats(pg, Server(c), Hours(c, 24), Rows(c, "limit", 25), as_of: AsOf(c)),
+            ["get_pg_replication_stats"] = (c, pg, an) => DarlingMcpPgReplicationStatsTools.GetPgReplicationStats(pg, Server(c), Hours(c, 24), Rows(c, "limit", 25), as_of: AsOf(c), cancellationToken: c.RequestAborted),
             ["get_pg_blocking"] = (c, pg, an) => DarlingMcpPgBlockingTools.GetPgBlocking(pg, Server(c), Hours(c, 24), Rows(c, "limit", 50), as_of: AsOf(c), cancellationToken: c.RequestAborted),
-            ["get_pg_database_stats"] = (c, pg, an) => DarlingMcpPgDatabaseTools.GetPgDatabaseStats(pg, Server(c), Hours(c, 24), Rows(c, "limit", 20), as_of: AsOf(c)),
-            ["get_pg_index_usage"] = (c, pg, an) => DarlingMcpPgIndexUsageTools.GetPgIndexUsage(pg, Server(c), Hours(c, 168), Rows(c, "limit", 25), as_of: AsOf(c)),
+            ["get_pg_database_stats"] = (c, pg, an) => DarlingMcpPgDatabaseTools.GetPgDatabaseStats(pg, Server(c), Hours(c, 24), Rows(c, "limit", 20), as_of: AsOf(c), cancellationToken: c.RequestAborted),
+            ["get_pg_index_usage"] = (c, pg, an) => DarlingMcpPgIndexUsageTools.GetPgIndexUsage(pg, Server(c), Hours(c, 168), Rows(c, "limit", 25), as_of: AsOf(c), cancellationToken: c.RequestAborted),
             ["get_pg_table_bloat"] = (c, pg, an) => DarlingMcpPgTableBloatTools.GetPgTableBloat(pg, Server(c), Hours(c, 168), Rows(c, "limit", 25), as_of: AsOf(c), cancellationToken: c.RequestAborted),
-            ["get_pg_session_states"] = (c, pg, an) => DarlingMcpPgSessionStatesTools.GetPgSessionStates(pg, Server(c), Hours(c, 24), Rows(c, "limit", 25), as_of: AsOf(c)),
+            ["get_pg_session_states"] = (c, pg, an) => DarlingMcpPgSessionStatesTools.GetPgSessionStates(pg, Server(c), Hours(c, 24), Rows(c, "limit", 25), as_of: AsOf(c), cancellationToken: c.RequestAborted),
             ["get_wait_stats"] = (c, pg, an) => DarlingMcpDataTools.GetWaitStats(pg, Server(c), Hours(c, 24), Rows(c, "limit", 20), as_of: AsOf(c), cancellationToken: c.RequestAborted),
             ["get_wait_trend"] = (c, pg, an) => RequireText(c, "wait_type", out var waitType)
                 ? (OptionalInt(c, "bucket_minutes", out var bucketMinutes)

Darling/PerformanceMonitor.Darling.Service/Mcp/DarlingMcpPgAutovacuumTools.cs

@@ -11,6 +11,7 @@
 using System.ComponentModel;
 using System.Linq;
 using System.Text.Json;
+using System.Threading;
 using System.Threading.Tasks;
 using ModelContextProtocol.Server;
 using Npgsql;
@@ -85,9 +86,10 @@ public static async Task<string> GetPgAutovacuumHealth(
         [Description("Server name or display name.")] string? server_name = null,
         [Description("Hours of history to analyze, used for the dead-tuple growth comparison. Default 24.")] int hours_back = 24,
         [Description("Maximum tables to return, worst first. Default 20. This is what bounds the page - read truncated to know whether more tables had pending work than were returned.")] int limit = 20,
-        [Description(McpHelpers.AsOfDescription)] string? as_of = null)
+        [Description(McpHelpers.AsOfDescription)] string? as_of = null,
+        CancellationToken cancellationToken = default)
     {
-        var (resolved, error) = await DarlingServerResolver.ResolveOrErrorAsync(postgres, server_name);
+        var (resolved, error) = await DarlingServerResolver.ResolveOrErrorAsync(postgres, server_name, cancellationToken);
         if (error != null) return error;
 
         var validation = McpHelpers.ValidateWindow(hours_back, as_of, out var windowEnd);
@@ -104,7 +106,7 @@ past the cap and McpHelpers.BoundPage turns that into (page, truncated) — the
                read a server with exactly `limit` tables behind on vacuum as a cut page and told the caller to
                raise a limit that had nothing more to give. */
             var fetched = await DarlingPgAutovacuumReader.GetPgAutovacuumAsync(
-                postgres, resolved.ServerId, now.AddHours(-hours_back), now, limit + 1);
+                postgres, resolved.ServerId, now.AddHours(-hours_back), now, limit + 1, cancellationToken);
             var (rows, truncated) = McpHelpers.BoundPage(fetched, limit);
 
             if (rows.Count == 0)
@@ -113,7 +115,7 @@ raise a limit that had nothing more to give. */
                    and a fabricated one for a SQL Server target — the collector has never run there. Ask the
                    engine before making the claim (#2532). */
                 var gated = await DarlingEngineCapability.NotCollectedStatusAsync(
-                    postgres, resolved.ServerId, resolved.ServerName, "pg_autovacuum_stats");
+                    postgres, resolved.ServerId, resolved.ServerName, "pg_autovacuum_stats", cancellationToken);
                 if (gated != null)
                 {
                     return gated;
@@ -139,7 +141,7 @@ a history that happened to be exactly RunsReadPerTable long. */
             var runs = await DarlingPgLogEventReader.GetAutovacuumRunsAsync(
                 postgres, resolved.ServerId, now.AddHours(-hours_back), now,
                 rows.Select(r => $"{r.SchemaName}.{r.TableName}").Distinct(StringComparer.Ordinal).ToList(),
-                RunsReadPerTable + 1);
+                RunsReadPerTable + 1, cancellationToken);
             var runsByTable = runs
                 .GroupBy(r => (Database: r.DatabaseName, Relation: r.RelationName))
                 .ToDictionary(g => g.Key, g => g.OrderByDescending(r => r.OccurredAtUtc).ToList());
@@ -265,7 +267,7 @@ reasons this read can and cannot separate. Said once here rather than per table.
                 tables,
             }, McpHelpers.JsonOptions);
         }
-        catch (Exception ex)
+        catch (Exception ex) when (ex is not OperationCanceledException)
         {
             return McpHelpers.FormatError("get_pg_autovacuum_health", ex);
         }

Darling/PerformanceMonitor.Darling.Service/Mcp/DarlingMcpPgDatabaseTools.cs

@@ -11,6 +11,7 @@
 using System.Globalization;
 using System.Linq;
 using System.Text.Json;
+using System.Threading;
 using System.Threading.Tasks;
 using ModelContextProtocol.Server;
 using Npgsql;
@@ -142,9 +143,10 @@ public static async Task<string> GetPgDatabaseStats(
         [Description("Server name or display name.")] string? server_name = null,
         [Description("Hours of history to analyze. Default 24.")] int hours_back = 24,
         [Description("Maximum databases to return, most temp bytes first. Default 20. Bounds databases[] only; read truncated to know whether more exist; totals stay of the whole window regardless.")] int limit = 20,
-        [Description(McpHelpers.AsOfDescription)] string? as_of = null)
+        [Description(McpHelpers.AsOfDescription)] string? as_of = null,
+        CancellationToken cancellationToken = default)
     {
-        var (resolved, error) = await DarlingServerResolver.ResolveOrErrorAsync(postgres, server_name);
+        var (resolved, error) = await DarlingServerResolver.ResolveOrErrorAsync(postgres, server_name, cancellationToken);
         if (error != null) return error;
 
         var validation = McpHelpers.ValidateWindow(hours_back, as_of, out var windowEnd);
@@ -160,16 +162,16 @@ public static async Task<string> GetPgDatabaseStats(
                truncation signal - and the window's totals ride on the same statement, so the figures below
                have a scope the cap cannot shrink. */
             var page = await DarlingPgDatabaseReader.GetPgDatabaseStatsPageAsync(
-                postgres, resolved.ServerId, start, end, limit + 1);
+                postgres, resolved.ServerId, start, end, limit + 1, cancellationToken);
 
             if (page.Rows.Count == 0)
             {
-                return await EmptyAsync(postgres, resolved.ServerId, resolved.ServerName, hours_back, start, end);
+                return await EmptyAsync(postgres, resolved.ServerId, resolved.ServerName, hours_back, start, end, cancellationToken);
             }
 
             return BuildDatabaseStatsJson(resolved.ServerName, hours_back, page, limit);
         }
-        catch (Exception ex)
+        catch (Exception ex) when (ex is not OperationCanceledException)
         {
             return McpHelpers.FormatError("get_pg_database_stats", ex);
         }
@@ -344,17 +346,18 @@ the statement (#3613 named it as the house rule's other arm). Kept beside cache_
     /// the next cycle to land.</para>
     /// </summary>
     private static async Task<string> EmptyAsync(
-        NpgsqlDataSource postgres, int serverId, string serverName, int hoursBack, DateTime start, DateTime end)
+        NpgsqlDataSource postgres, int serverId, string serverName, int hoursBack, DateTime start, DateTime end,
+        CancellationToken cancellationToken = default)
     {
         var gated = await DarlingEngineCapability.NotCollectedStatusAsync(
-            postgres, serverId, serverName, "pg_database_stats");
+            postgres, serverId, serverName, "pg_database_stats", cancellationToken);
         if (gated != null)
         {
             return gated;
         }
 
         var (samplesInWindow, everCollected) = await DarlingPgDatabaseReader.GetCoverageAsync(
-            postgres, serverId, start, end);
+            postgres, serverId, start, end, cancellationToken);
 
         var hints = new
         {

Darling/PerformanceMonitor.Darling.Service/Mcp/DarlingMcpPgIndexUsageTools.cs

@@ -12,6 +12,7 @@
 using System.Globalization;
 using System.Linq;
 using System.Text.Json;
+using System.Threading;
 using System.Threading.Tasks;
 using ModelContextProtocol.Server;
 using Npgsql;
@@ -228,9 +229,10 @@ public static async Task<string> GetPgIndexUsage(
         [Description("Server name or display name.")] string? server_name = null,
         [Description("Hours of history to analyze. Default 168 (seven days). Widen this to the longest interval any scheduled job runs on before calling an index unused.")] int hours_back = 168,
         [Description("Maximum indexes to return, biggest unscanned first. Default 25. See the tool's reading guide.")] int limit = 25,
-        [Description(McpHelpers.AsOfDescription)] string? as_of = null)
+        [Description(McpHelpers.AsOfDescription)] string? as_of = null,
+        CancellationToken cancellationToken = default)
     {
-        var (resolved, error) = await DarlingServerResolver.ResolveOrErrorAsync(postgres, server_name);
+        var (resolved, error) = await DarlingServerResolver.ResolveOrErrorAsync(postgres, server_name, cancellationToken);
         if (error != null) return error;
 
         var validation = McpHelpers.ValidateWindow(hours_back, as_of, out var windowEnd);
@@ -247,12 +249,12 @@ public static async Task<string> GetPgIndexUsage(
                exactly `limit` reportable indexes as a cut page. Bound BEFORE the aggregates below, so every count
                is a count of the page. */
             var fetched = await DarlingPgIndexUsageReader.GetPgIndexUsageAsync(
-                postgres, resolved.ServerId, start, end, limit + 1);
+                postgres, resolved.ServerId, start, end, limit + 1, cancellationToken);
             var (rows, truncated) = McpHelpers.BoundPage(fetched, limit);
 
             if (rows.Count == 0)
             {
-                return await EmptyAsync(postgres, resolved.ServerId, resolved.ServerName, hours_back, start, end);
+                return await EmptyAsync(postgres, resolved.ServerId, resolved.ServerName, hours_back, start, end, cancellationToken);
             }
 
             var unscanned = rows.Where(r => r.ScansInWindow == 0 && r.IsValid && !r.IsPrimaryKey
@@ -366,7 +368,7 @@ field name has to survive being read by something that will act on it. */
                 indexes,
             }, McpHelpers.JsonOptions);
         }
-        catch (Exception ex)
+        catch (Exception ex) when (ex is not OperationCanceledException)
         {
             return McpHelpers.FormatError("get_pg_index_usage", ex);
         }
@@ -382,16 +384,17 @@ field name has to survive being read by something that will act on it. */
     /// windowed scan count is a difference and needs two.</para>
     /// </summary>
     private static async Task<string> EmptyAsync(
-        NpgsqlDataSource postgres, int serverId, string serverName, int hoursBack, DateTime start, DateTime end)
+        NpgsqlDataSource postgres, int serverId, string serverName, int hoursBack, DateTime start, DateTime end,
+        CancellationToken cancellationToken = default)
     {
         var gated = await DarlingEngineCapability.NotCollectedStatusAsync(
-            postgres, serverId, serverName, "pg_index_usage_stats");
+            postgres, serverId, serverName, "pg_index_usage_stats", cancellationToken);
         if (gated != null)
         {
             return gated;
         }
 
-        var probe = await DarlingPgIndexUsageReader.ProbePgIndexUsageAsync(postgres, serverId, start, end);
+        var probe = await DarlingPgIndexUsageReader.ProbePgIndexUsageAsync(postgres, serverId, start, end, cancellationToken);
 
         var hints = new
         {