From 575ea4fbf0fb2260b4b10f7ebb48de3fa8ddd567 Mon Sep 17 00:00:00 2001 From: Erik Darling <2136037+erikdarlingdata@users.noreply.github.com> Date: Fri, 2 Oct 2026 23:49:59 -0400 Subject: [PATCH 1/5] Read the Query Store chunk floors once per fleet eligibility check (#4605) --- ...QueryStoreWideFleetEligibilityLiveTests.cs | 182 ++++++++++++++++++ .../DarlingWebEndpoints.cs | 59 +++++- .../QueryStoreIntervalWide.cs | 45 ++++- 3 files changed, 277 insertions(+), 9 deletions(-) create mode 100644 Darling/Darling.Tests/QueryStoreWideFleetEligibilityLiveTests.cs diff --git a/Darling/Darling.Tests/QueryStoreWideFleetEligibilityLiveTests.cs b/Darling/Darling.Tests/QueryStoreWideFleetEligibilityLiveTests.cs new file mode 100644 index 0000000000..9410fddb59 --- /dev/null +++ b/Darling/Darling.Tests/QueryStoreWideFleetEligibilityLiveTests.cs @@ -0,0 +1,182 @@ +/* + * Copyright (c) 2026 Erik Darling, Darling Data LLC + * + * This file is part of the SQL Server Performance Monitor. + * + * Licensed under the MIT License. See LICENSE file in the project root for full license information. + */ + +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Diagnostics; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Npgsql; +using PerformanceMonitor.Collectors; +using PerformanceMonitor.Darling.Service; +using PerformanceMonitor.Darling.Storage; +using Xunit; + +namespace Darling.Tests; + +/// +/// #4605: a fleet Custom Views panel's table-eligibility check reads the store-wide chunk floors once, not once +/// per server, and answers exactly what checking each server one after another answers. +/// +/* #1776 own-store: deliberately NOT [Collection("live-postgres")]; works entirely inside its own scratch database. */ +public sealed class QueryStoreWideFleetEligibilityLiveTests +{ + private static readonly DateTime S = DateTime.SpecifyKind(DateTime.UtcNow.Date.AddDays(-6), DateTimeKind.Unspecified); + + private static async Task<(ScratchPostgres Scratch, NpgsqlDataSource Postgres, NpgsqlConnection Connection)> SeedAsync( + string baseCs, int[] eligibleIds, int? legacyId, CancellationToken ct) + { + var scratch = await ScratchPostgres.CreateAsync(baseCs, ct); + var connection = new NpgsqlConnection(scratch.ConnectionString); + await connection.OpenAsync(ct); + await PgMigrations.MigrateAsync(connection, ct); + Assert.True(await TimescaleSupport.TryEnableAsync(connection, null, ct), "TimescaleDB must be enabled on the test cluster"); + await TimescaleSupport.ConvertToHypertablesAsync(connection, null, ct); + var postgres = NpgsqlDataSource.Create(scratch.ConnectionString); + var runner = new DarlingCollectorRunner(postgres, new CollectorDeltaCalculator()); + var all = eligibleIds.Concat(legacyId is int l ? new[] { l } : Array.Empty()).ToArray(); + for (var i = 0; i < all.Length; i++) + { + var id = all[i]; + if (id == legacyId) + { + await QueryStoreIntervalWideGridLiveTests.SeedGridLegacyAsync(runner, id, S, ct); + } + else + { + await QueryStoreIntervalWideGridLiveTests.SeedGridAsync(runner, id, S, ct); + } + + await using var ins = new NpgsqlCommand( + "INSERT INTO servers (server_id, server_name, display_name, is_enabled, sql_major_version, created_date, modified_date) " + + $"VALUES ({id}, 'qsiw-fleet-{i:00}', 'qsiw-fleet-{i:00}', TRUE, 16, now(), now()) ON CONFLICT (server_id) DO UPDATE SET is_enabled = TRUE", connection); + await ins.ExecuteNonQueryAsync(ct); + await QueryStoreIntervalWideBelowFloorLiveTests.SeedQueryStoreLogAsync(connection, id, S.AddDays(-60), S.AddDays(4), null, null, ct); + } + + /* Different read starts: servers 0 and 3 share the latest claim, so the tie goes to server 0. */ + for (var i = 0; i < eligibleIds.Length; i++) + { + var hours = i == 0 || i == 3 ? 14 : 12 + (i % 2); + await using var upd = new NpgsqlCommand( + $"UPDATE collect.query_store_interval_wide_coverage SET filled_since = TIMESTAMP '{S.AddHours(hours):yyyy-MM-dd HH:mm:ss}' WHERE server_id = {eligibleIds[i]}", connection); + await upd.ExecuteNonQueryAsync(ct); + } + + await using var drop = new NpgsqlCommand( + $"SELECT drop_chunks('collect.query_store_stats', older_than => TIMESTAMP '{S.AddDays(1):yyyy-MM-dd HH:mm:ss}')", connection); + await drop.ExecuteNonQueryAsync(ct); + return (scratch, postgres, connection); + } + + private static async Task EndAsync(NpgsqlConnection connection, int id, CancellationToken ct) + { + await using var cmd = new NpgsqlCommand($"SELECT MAX(applied_through) FROM collect.query_store_interval_wide_coverage", connection); + return (DateTime)(await cmd.ExecuteScalarAsync(ct))!; + } + + /// The pre-change answer: each server resolved one after another, re-reading the floors every time. + private static async Task<(bool Eligible, DateTime? WideStart, QueryStoreIntervalWide.WideStartBound Bound, string? SettingServer)> SerialAsync( + NpgsqlConnection connection, DateTime end, CancellationToken ct) + { + var servers = new List<(int Id, string Name)>(); + await using (var cmd = new NpgsqlCommand("SELECT server_id, server_name FROM collect.servers WHERE is_enabled", connection)) + await using (var reader = await cmd.ExecuteReaderAsync(ct)) + { + while (await reader.ReadAsync(ct)) + { + servers.Add((reader.GetInt32(0), reader.GetString(1))); + } + } + + var wideStart = S; + var bound = QueryStoreIntervalWide.WideStartBound.Window; + string? setting = null; + foreach (var (id, name) in servers) + { + var plan = await QueryStoreIntervalWide.ResolveReadAsync(connection, id, S, end, end, TimeSpan.FromHours(12), 60, null, ct); + if (!plan.UseTable) + { + return default; + } + + if (plan.ReadStart > wideStart) + { + wideStart = plan.ReadStart; + bound = plan.StartBound; + setting = name; + } + } + + return (true, wideStart, bound, setting); + } + + [Fact] + public async Task FleetCheck_ReadsTheStoreWideFloorsOnce_AndEqualsTheSerialAnswer_AcrossRepeatedRuns() + { + var baseCs = Environment.GetEnvironmentVariable("DARLING_TEST_PG"); + Assert.SkipWhen(string.IsNullOrEmpty(baseCs), "Set DARLING_TEST_PG to a Postgres connection string to run the #4605 live test."); + var ct = TestContext.Current.CancellationToken; + var ids = new[] { -4605101, -4605102, -4605103, -4605104, -4605105, -4605106 }; + var (scratch, _, connection) = await SeedAsync(baseCs!, ids, null, ct); + await using var _s = scratch; + await using var _c = connection; + var end = await EndAsync(connection, ids[0], ct); + + var expected = await SerialAsync(connection, end, ct); + Assert.True(expected.Eligible); + Assert.Equal(S.AddHours(14), expected.WideStart); + Assert.Equal("qsiw-fleet-00", expected.SettingServer); + + var floorReads = 0; + using var listener = new ActivityListener + { + ShouldListenTo = src => src.Name == "Npgsql", + Sample = (ref ActivityCreationOptions _) => ActivitySamplingResult.AllDataAndRecorded, + ActivityStopped = a => + { + var text = (a.GetTagItem("db.query.text") ?? a.GetTagItem("db.statement")) as string; + if (text is not null && text.Contains("table_is_hypertable", StringComparison.Ordinal) && text.Contains("raw_floor", StringComparison.Ordinal)) + { + Interlocked.Increment(ref floorReads); + } + }, + }; + ActivitySource.AddActivityListener(listener); + + await using var postgres = NpgsqlDataSource.Create(scratch.ConnectionString); + for (var run = 0; run < 20; run++) + { + floorReads = 0; + var got = await DarlingWebEndpoints.ResolveQueryStoreWideEligibleAsync(postgres, null, S, end, end, ct); + Assert.Equal(expected, got); + Assert.Equal(1, floorReads); + } + } + + [Fact] + public async Task FleetCheck_WithOneServerHoldingLegacyRows_ReadsRaw_LikeTheSerialAnswer() + { + var baseCs = Environment.GetEnvironmentVariable("DARLING_TEST_PG"); + Assert.SkipWhen(string.IsNullOrEmpty(baseCs), "Set DARLING_TEST_PG to a Postgres connection string to run the #4605 live test."); + var ct = TestContext.Current.CancellationToken; + var ids = new[] { -4605201, -4605202, -4605203 }; + var (scratch, postgres, connection) = await SeedAsync(baseCs!, ids, -4605204, ct); + await using var _s = scratch; + await using var _c = connection; + await using var _p = postgres; + var end = await EndAsync(connection, ids[0], ct); + + Assert.Equal(default, await SerialAsync(connection, end, ct)); + var got = await DarlingWebEndpoints.ResolveQueryStoreWideEligibleAsync(postgres, null, S, end, end, ct); + Assert.False(got.Eligible); + Assert.Null(got.WideStart); + } +} diff --git a/Darling/PerformanceMonitor.Darling.Service/DarlingWebEndpoints.cs b/Darling/PerformanceMonitor.Darling.Service/DarlingWebEndpoints.cs index d255c6d32d..c3a9c53305 100644 --- a/Darling/PerformanceMonitor.Darling.Service/DarlingWebEndpoints.cs +++ b/Darling/PerformanceMonitor.Darling.Service/DarlingWebEndpoints.cs @@ -1476,6 +1476,9 @@ so a test can run it without a live Postgres round trip. */ private const string QueryStoreWideServerIdsSql = "SELECT server_id, server_name FROM collect.servers WHERE is_enabled AND ($1::text[] IS NULL OR server_name = ANY($1))"; + /// #4605: how many servers' eligibility probes run at once in a fleet check. + private const int QueryStoreWideEligibilityConcurrency = 4; + /// /// #4605: whether a composed Query Store panel over .. /// may read collect.query_store_interval_wide (V145) instead of deduping raw — decided here, in the @@ -1538,21 +1541,63 @@ so a test can run it without a live Postgres round trip. */ var wideStart = start; var bound = QueryStoreIntervalWide.WideStartBound.Window; string? settingServer = null; - foreach (var (serverId, serverName) in wideServers) + + /* The floors are one store-wide answer: read once here, not once per server. Servers are then + checked QueryStoreWideEligibilityConcurrency at a time, each on its own pooled connection; the + first refusal cancels the rest. The reduction below walks wideServers in order, so ties for the + latest read start go to the earliest server exactly as a serial loop would. */ + var storeWide = await QueryStoreIntervalWide.ReadStoreWideInputsAsync(connection, McpCommandDeadlines.ReadSeconds, cancellationToken); + using var refused = System.Threading.CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + using var gate = new System.Threading.SemaphoreSlim(QueryStoreWideEligibilityConcurrency); + var checks = wideServers.Select(async server => { - var plan = await QueryStoreIntervalWide.ResolveReadAsync( - connection, serverId, start, end, literalWindowEnd, ComposeQueryStoreWideMinWindow, - McpCommandDeadlines.ReadSeconds, logger: null, cancellationToken); - if (!plan.UseTable) + try { - return default; + await gate.WaitAsync(refused.Token); + } + catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested) + { + return (QueryStoreIntervalWide.WideReadPlan?)null; } + try + { + await using var serverConnection = await postgres.OpenConnectionAsync(refused.Token); + var plan = await QueryStoreIntervalWide.ResolveReadAsync( + serverConnection, server.Id, start, end, literalWindowEnd, ComposeQueryStoreWideMinWindow, + McpCommandDeadlines.ReadSeconds, logger: null, refused.Token, storeWide); + if (!plan.UseTable) + { + refused.Cancel(); + } + + return (QueryStoreIntervalWide.WideReadPlan?)plan; + } + catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested) + { + return null; + } + finally + { + gate.Release(); + } + }).ToList(); + var plans = await Task.WhenAll(checks); + cancellationToken.ThrowIfCancellationRequested(); + + if (refused.IsCancellationRequested) + { + return default; + } + + for (var i = 0; i < wideServers.Count; i++) + { + var plan = plans[i]!.Value; if (plan.ReadStart > wideStart) { wideStart = plan.ReadStart; bound = plan.StartBound; - settingServer = serverName; + settingServer = wideServers[i].Name; } } diff --git a/Darling/PerformanceMonitor.Darling.Storage/QueryStoreIntervalWide.cs b/Darling/PerformanceMonitor.Darling.Storage/QueryStoreIntervalWide.cs index 1d924bc947..3bc4d61480 100644 --- a/Darling/PerformanceMonitor.Darling.Storage/QueryStoreIntervalWide.cs +++ b/Darling/PerformanceMonitor.Darling.Storage/QueryStoreIntervalWide.cs @@ -973,6 +973,40 @@ public static (DateTime Start, WideStartBound Bound)? ExactBelowFloorStart( return (p.UseTable, p.ClampedStart); } + /// + /// The inputs reads that do not depend on the server: the three floors of + /// , which takes no parameter and reads TimescaleDB's catalog for the whole store. + /// ('s coverage and pending columns, , + /// and the cadence probe all filter on the server, so they stay per server.) + /// A caller that resolves many servers for one window reads this once and passes it to each call. + /// + public sealed record StoreWideInputs(DateTime? RawFloor, bool TableIsHypertable, DateTime? TableFloor); + + /// Reads once. Null when TimescaleDB's catalog is absent, which + /// is the answer would reach itself (it never reads the floors then). + public static async Task ReadStoreWideInputsAsync( + NpgsqlConnection connection, int commandTimeoutSeconds, CancellationToken cancellationToken) + { + bool hasTimescale; + await using (var probe = new NpgsqlCommand("SELECT to_regclass('timescaledb_information.chunks') IS NOT NULL;", connection) { CommandTimeout = commandTimeoutSeconds }) + { + hasTimescale = (bool)(await probe.ExecuteScalarAsync(cancellationToken))!; + } + + if (!hasTimescale) + { + return null; + } + + await using var floors = new NpgsqlCommand(ChunkFloorsSql, connection) { CommandTimeout = commandTimeoutSeconds }; + await using var reader = await floors.ExecuteReaderAsync(cancellationToken); + await reader.ReadAsync(cancellationToken); + return new StoreWideInputs( + reader.IsDBNull(0) ? null : reader.GetDateTime(0), + reader.GetBoolean(1), + reader.IsDBNull(2) ? null : reader.GetDateTime(2)); + } + /// /// plus the store round trips it needs, the clamp (), and the /// lower bound the table read may bind (), which reaches below raw's chunk @@ -988,7 +1022,8 @@ public static async Task ResolveReadAsync( TimeSpan minWindow, int commandTimeoutSeconds, ILogger? logger, - CancellationToken cancellationToken) + CancellationToken cancellationToken, + StoreWideInputs? storeWide = null) { /* The probes bind these as timestamp-without-time-zone, which Npgsql refuses for Kind=Utc; a caller that resolves its window from DateTime.UtcNow (the MCP tool) would otherwise fail every decision and read raw. */ @@ -1029,7 +1064,13 @@ remaining round trips (ChunkFloorsSql's metadata read and, worse, PlainTableFloo DateTime? rawFloor = null; DateTime? tableFloor = null; var tableIsHypertable = false; - if (hasTimescale) + if (hasTimescale && storeWide is not null) + { + rawFloor = storeWide.RawFloor; + tableIsHypertable = storeWide.TableIsHypertable; + tableFloor = storeWide.TableFloor; + } + else if (hasTimescale) { await using var floors = new NpgsqlCommand(ChunkFloorsSql, connection) { CommandTimeout = commandTimeoutSeconds }; await using var reader = await floors.ExecuteReaderAsync(cancellationToken); From beebcb98ad4d90fbbddb1133be9d6c648146c48e Mon Sep 17 00:00:00 2001 From: Erik Darling <2136037+erikdarlingdata@users.noreply.github.com> Date: Fri, 2 Oct 2026 23:59:27 -0400 Subject: [PATCH 2/5] Order the fleet eligibility server list, scope the statement pin to its scratch database, close the connection before the fan-out (#4605) --- ...QueryStoreWideFleetEligibilityLiveTests.cs | 12 +++- .../DarlingWebEndpoints.cs | 56 ++++++++++--------- 2 files changed, 40 insertions(+), 28 deletions(-) diff --git a/Darling/Darling.Tests/QueryStoreWideFleetEligibilityLiveTests.cs b/Darling/Darling.Tests/QueryStoreWideFleetEligibilityLiveTests.cs index 9410fddb59..837d21e163 100644 --- a/Darling/Darling.Tests/QueryStoreWideFleetEligibilityLiveTests.cs +++ b/Darling/Darling.Tests/QueryStoreWideFleetEligibilityLiveTests.cs @@ -61,7 +61,7 @@ public sealed class QueryStoreWideFleetEligibilityLiveTests await QueryStoreIntervalWideBelowFloorLiveTests.SeedQueryStoreLogAsync(connection, id, S.AddDays(-60), S.AddDays(4), null, null, ct); } - /* Different read starts: servers 0 and 3 share the latest claim, so the tie goes to server 0. */ + /* Different read starts: servers 0 and 3 share the latest claim, so the tie goes to the lower server_id, qsiw-fleet-03. */ for (var i = 0; i < eligibleIds.Length; i++) { var hours = i == 0 || i == 3 ? 14 : 12 + (i % 2); @@ -87,7 +87,7 @@ private static async Task EndAsync(NpgsqlConnection connection, int id NpgsqlConnection connection, DateTime end, CancellationToken ct) { var servers = new List<(int Id, string Name)>(); - await using (var cmd = new NpgsqlCommand("SELECT server_id, server_name FROM collect.servers WHERE is_enabled", connection)) + await using (var cmd = new NpgsqlCommand("SELECT server_id, server_name FROM collect.servers WHERE is_enabled ORDER BY server_id", connection)) await using (var reader = await cmd.ExecuteReaderAsync(ct)) { while (await reader.ReadAsync(ct)) @@ -133,15 +133,21 @@ public async Task FleetCheck_ReadsTheStoreWideFloorsOnce_AndEqualsTheSerialAnswe var expected = await SerialAsync(connection, end, ct); Assert.True(expected.Eligible); Assert.Equal(S.AddHours(14), expected.WideStart); - Assert.Equal("qsiw-fleet-00", expected.SettingServer); + Assert.Equal("qsiw-fleet-03", expected.SettingServer); var floorReads = 0; + var scratchDb = new NpgsqlConnectionStringBuilder(scratch.ConnectionString).Database; using var listener = new ActivityListener { ShouldListenTo = src => src.Name == "Npgsql", Sample = (ref ActivityCreationOptions _) => ActivitySamplingResult.AllDataAndRecorded, ActivityStopped = a => { + if (!string.Equals(a.GetTagItem("db.namespace") as string, scratchDb, StringComparison.Ordinal)) + { + return; + } + var text = (a.GetTagItem("db.query.text") ?? a.GetTagItem("db.statement")) as string; if (text is not null && text.Contains("table_is_hypertable", StringComparison.Ordinal) && text.Contains("raw_floor", StringComparison.Ordinal)) { diff --git a/Darling/PerformanceMonitor.Darling.Service/DarlingWebEndpoints.cs b/Darling/PerformanceMonitor.Darling.Service/DarlingWebEndpoints.cs index c3a9c53305..7bb03345a5 100644 --- a/Darling/PerformanceMonitor.Darling.Service/DarlingWebEndpoints.cs +++ b/Darling/PerformanceMonitor.Darling.Service/DarlingWebEndpoints.cs @@ -1474,7 +1474,7 @@ so a test can run it without a live Postgres round trip. */ /// #4617: see . private const string QueryStoreWideServerIdsSql = - "SELECT server_id, server_name FROM collect.servers WHERE is_enabled AND ($1::text[] IS NULL OR server_name = ANY($1))"; + "SELECT server_id, server_name FROM collect.servers WHERE is_enabled AND ($1::text[] IS NULL OR server_name = ANY($1)) ORDER BY server_id"; /// #4605: how many servers' eligibility probes run at once in a fleet check. private const int QueryStoreWideEligibilityConcurrency = 4; @@ -1502,37 +1502,44 @@ so a test can run it without a live Postgres round trip. */ try { - await using var connection = await postgres.OpenConnectionAsync(cancellationToken); + var wideServers = new List<(int Id, string Name)>(); + QueryStoreIntervalWide.StoreWideInputs? storeWide; - int schemaVersion; - await using (var probe = new NpgsqlCommand(QueryStoreWideSchemaVersionSql, connection) { CommandTimeout = McpCommandDeadlines.ReadSeconds }) + /* The schema version, the server list and the floors are read on one connection that is closed + before the fan-out, so a fleet check holds at most QueryStoreWideEligibilityConcurrency connections. */ + await using (var connection = await postgres.OpenConnectionAsync(cancellationToken)) { - schemaVersion = (int)(await probe.ExecuteScalarAsync(cancellationToken))!; - } + int schemaVersion; + await using (var probe = new NpgsqlCommand(QueryStoreWideSchemaVersionSql, connection) { CommandTimeout = McpCommandDeadlines.ReadSeconds }) + { + schemaVersion = (int)(await probe.ExecuteScalarAsync(cancellationToken))!; + } - if (schemaVersion < 145) - { - return default; - } + if (schemaVersion < 145) + { + return default; + } - var wideServers = new List<(int Id, string Name)>(); - await using (var servers = new NpgsqlCommand(QueryStoreWideServerIdsSql, connection) { CommandTimeout = McpCommandDeadlines.ReadSeconds }) - { - servers.Parameters.Add(new NpgsqlParameter + await using (var servers = new NpgsqlCommand(QueryStoreWideServerIdsSql, connection) { CommandTimeout = McpCommandDeadlines.ReadSeconds }) { - NpgsqlDbType = NpgsqlTypes.NpgsqlDbType.Array | NpgsqlTypes.NpgsqlDbType.Text, - Value = serverScope is { Count: > 0 } ? (object)serverScope.ToArray() : DBNull.Value, - }); - await using var reader = await servers.ExecuteReaderAsync(cancellationToken); - while (await reader.ReadAsync(cancellationToken)) + servers.Parameters.Add(new NpgsqlParameter + { + NpgsqlDbType = NpgsqlTypes.NpgsqlDbType.Array | NpgsqlTypes.NpgsqlDbType.Text, + Value = serverScope is { Count: > 0 } ? (object)serverScope.ToArray() : DBNull.Value, + }); + await using var reader = await servers.ExecuteReaderAsync(cancellationToken); + while (await reader.ReadAsync(cancellationToken)) + { + wideServers.Add((reader.GetInt32(0), reader.GetString(1))); + } + } + + if (wideServers.Count == 0) { - wideServers.Add((reader.GetInt32(0), reader.GetString(1))); + return default; } - } - if (wideServers.Count == 0) - { - return default; + storeWide = await QueryStoreIntervalWide.ReadStoreWideInputsAsync(connection, McpCommandDeadlines.ReadSeconds, cancellationToken); } /* #4689: every server in scope reads from ONE common start, the latest of the per-server read @@ -1546,7 +1553,6 @@ so a test can run it without a live Postgres round trip. */ checked QueryStoreWideEligibilityConcurrency at a time, each on its own pooled connection; the first refusal cancels the rest. The reduction below walks wideServers in order, so ties for the latest read start go to the earliest server exactly as a serial loop would. */ - var storeWide = await QueryStoreIntervalWide.ReadStoreWideInputsAsync(connection, McpCommandDeadlines.ReadSeconds, cancellationToken); using var refused = System.Threading.CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); using var gate = new System.Threading.SemaphoreSlim(QueryStoreWideEligibilityConcurrency); var checks = wideServers.Select(async server => From cc94f7faec147e925cfc4fee4b4af202b0da7263 Mon Sep 17 00:00:00 2001 From: Erik Darling <2136037+erikdarlingdata@users.noreply.github.com> Date: Sat, 3 Oct 2026 00:20:35 -0400 Subject: [PATCH 3/5] Check fleet eligibility serially again; keep the floors read once (#4605) --- .../DarlingWebEndpoints.cs | 112 +++++------------- 1 file changed, 32 insertions(+), 80 deletions(-) diff --git a/Darling/PerformanceMonitor.Darling.Service/DarlingWebEndpoints.cs b/Darling/PerformanceMonitor.Darling.Service/DarlingWebEndpoints.cs index 7bb03345a5..8396d91884 100644 --- a/Darling/PerformanceMonitor.Darling.Service/DarlingWebEndpoints.cs +++ b/Darling/PerformanceMonitor.Darling.Service/DarlingWebEndpoints.cs @@ -1476,9 +1476,6 @@ so a test can run it without a live Postgres round trip. */ private const string QueryStoreWideServerIdsSql = "SELECT server_id, server_name FROM collect.servers WHERE is_enabled AND ($1::text[] IS NULL OR server_name = ANY($1)) ORDER BY server_id"; - /// #4605: how many servers' eligibility probes run at once in a fleet check. - private const int QueryStoreWideEligibilityConcurrency = 4; - /// /// #4605: whether a composed Query Store panel over .. /// may read collect.query_store_interval_wide (V145) instead of deduping raw — decided here, in the @@ -1502,44 +1499,37 @@ so a test can run it without a live Postgres round trip. */ try { - var wideServers = new List<(int Id, string Name)>(); - QueryStoreIntervalWide.StoreWideInputs? storeWide; + await using var connection = await postgres.OpenConnectionAsync(cancellationToken); - /* The schema version, the server list and the floors are read on one connection that is closed - before the fan-out, so a fleet check holds at most QueryStoreWideEligibilityConcurrency connections. */ - await using (var connection = await postgres.OpenConnectionAsync(cancellationToken)) + int schemaVersion; + await using (var probe = new NpgsqlCommand(QueryStoreWideSchemaVersionSql, connection) { CommandTimeout = McpCommandDeadlines.ReadSeconds }) { - int schemaVersion; - await using (var probe = new NpgsqlCommand(QueryStoreWideSchemaVersionSql, connection) { CommandTimeout = McpCommandDeadlines.ReadSeconds }) - { - schemaVersion = (int)(await probe.ExecuteScalarAsync(cancellationToken))!; - } + schemaVersion = (int)(await probe.ExecuteScalarAsync(cancellationToken))!; + } - if (schemaVersion < 145) - { - return default; - } + if (schemaVersion < 145) + { + return default; + } - await using (var servers = new NpgsqlCommand(QueryStoreWideServerIdsSql, connection) { CommandTimeout = McpCommandDeadlines.ReadSeconds }) + var wideServers = new List<(int Id, string Name)>(); + await using (var servers = new NpgsqlCommand(QueryStoreWideServerIdsSql, connection) { CommandTimeout = McpCommandDeadlines.ReadSeconds }) + { + servers.Parameters.Add(new NpgsqlParameter { - servers.Parameters.Add(new NpgsqlParameter - { - NpgsqlDbType = NpgsqlTypes.NpgsqlDbType.Array | NpgsqlTypes.NpgsqlDbType.Text, - Value = serverScope is { Count: > 0 } ? (object)serverScope.ToArray() : DBNull.Value, - }); - await using var reader = await servers.ExecuteReaderAsync(cancellationToken); - while (await reader.ReadAsync(cancellationToken)) - { - wideServers.Add((reader.GetInt32(0), reader.GetString(1))); - } - } - - if (wideServers.Count == 0) + NpgsqlDbType = NpgsqlTypes.NpgsqlDbType.Array | NpgsqlTypes.NpgsqlDbType.Text, + Value = serverScope is { Count: > 0 } ? (object)serverScope.ToArray() : DBNull.Value, + }); + await using var reader = await servers.ExecuteReaderAsync(cancellationToken); + while (await reader.ReadAsync(cancellationToken)) { - return default; + wideServers.Add((reader.GetInt32(0), reader.GetString(1))); } + } - storeWide = await QueryStoreIntervalWide.ReadStoreWideInputsAsync(connection, McpCommandDeadlines.ReadSeconds, cancellationToken); + if (wideServers.Count == 0) + { + return default; } /* #4689: every server in scope reads from ONE common start, the latest of the per-server read @@ -1549,61 +1539,23 @@ so a test can run it without a live Postgres round trip. */ var bound = QueryStoreIntervalWide.WideStartBound.Window; string? settingServer = null; - /* The floors are one store-wide answer: read once here, not once per server. Servers are then - checked QueryStoreWideEligibilityConcurrency at a time, each on its own pooled connection; the - first refusal cancels the rest. The reduction below walks wideServers in order, so ties for the - latest read start go to the earliest server exactly as a serial loop would. */ - using var refused = System.Threading.CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); - using var gate = new System.Threading.SemaphoreSlim(QueryStoreWideEligibilityConcurrency); - var checks = wideServers.Select(async server => + /* The floors are one store-wide answer: read once here, not once per server. */ + var storeWide = await QueryStoreIntervalWide.ReadStoreWideInputsAsync(connection, McpCommandDeadlines.ReadSeconds, cancellationToken); + foreach (var (serverId, serverName) in wideServers) { - try - { - await gate.WaitAsync(refused.Token); - } - catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested) + var plan = await QueryStoreIntervalWide.ResolveReadAsync( + connection, serverId, start, end, literalWindowEnd, ComposeQueryStoreWideMinWindow, + McpCommandDeadlines.ReadSeconds, logger: null, cancellationToken, storeWide); + if (!plan.UseTable) { - return (QueryStoreIntervalWide.WideReadPlan?)null; - } - - try - { - await using var serverConnection = await postgres.OpenConnectionAsync(refused.Token); - var plan = await QueryStoreIntervalWide.ResolveReadAsync( - serverConnection, server.Id, start, end, literalWindowEnd, ComposeQueryStoreWideMinWindow, - McpCommandDeadlines.ReadSeconds, logger: null, refused.Token, storeWide); - if (!plan.UseTable) - { - refused.Cancel(); - } - - return (QueryStoreIntervalWide.WideReadPlan?)plan; - } - catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested) - { - return null; - } - finally - { - gate.Release(); + return default; } - }).ToList(); - var plans = await Task.WhenAll(checks); - cancellationToken.ThrowIfCancellationRequested(); - if (refused.IsCancellationRequested) - { - return default; - } - - for (var i = 0; i < wideServers.Count; i++) - { - var plan = plans[i]!.Value; if (plan.ReadStart > wideStart) { wideStart = plan.ReadStart; bound = plan.StartBound; - settingServer = wideServers[i].Name; + settingServer = serverName; } } From 964bf56926c63b31c3c73f3b81cb925ae9a152de Mon Sep 17 00:00:00 2001 From: Erik Darling <2136037+erikdarlingdata@users.noreply.github.com> Date: Sat, 3 Oct 2026 00:53:41 -0400 Subject: [PATCH 4/5] Read the fleet check's chunk floors lazily through a per-check cache (#4605) --- ...QueryStoreWideFleetEligibilityLiveTests.cs | 103 ++++++++++++------ .../DarlingWebEndpoints.cs | 5 +- .../QueryStoreIntervalWide.cs | 45 +++----- 3 files changed, 91 insertions(+), 62 deletions(-) diff --git a/Darling/Darling.Tests/QueryStoreWideFleetEligibilityLiveTests.cs b/Darling/Darling.Tests/QueryStoreWideFleetEligibilityLiveTests.cs index 837d21e163..209e8f6f18 100644 --- a/Darling/Darling.Tests/QueryStoreWideFleetEligibilityLiveTests.cs +++ b/Darling/Darling.Tests/QueryStoreWideFleetEligibilityLiveTests.cs @@ -9,7 +9,6 @@ using System; using System.Collections.Concurrent; using System.Collections.Generic; -using System.Diagnostics; using System.Linq; using System.Threading; using System.Threading.Tasks; @@ -34,12 +33,16 @@ public sealed class QueryStoreWideFleetEligibilityLiveTests string baseCs, int[] eligibleIds, int? legacyId, CancellationToken ct) { var scratch = await ScratchPostgres.CreateAsync(baseCs, ct); - var connection = new NpgsqlConnection(scratch.ConnectionString); + NpgsqlConnection? connection = null; + NpgsqlDataSource? postgres = null; + try + { + connection = new NpgsqlConnection(scratch.ConnectionString); await connection.OpenAsync(ct); await PgMigrations.MigrateAsync(connection, ct); Assert.True(await TimescaleSupport.TryEnableAsync(connection, null, ct), "TimescaleDB must be enabled on the test cluster"); await TimescaleSupport.ConvertToHypertablesAsync(connection, null, ct); - var postgres = NpgsqlDataSource.Create(scratch.ConnectionString); + postgres = NpgsqlDataSource.Create(scratch.ConnectionString); var runner = new DarlingCollectorRunner(postgres, new CollectorDeltaCalculator()); var all = eligibleIds.Concat(legacyId is int l ? new[] { l } : Array.Empty()).ToArray(); for (var i = 0; i < all.Length; i++) @@ -74,11 +77,27 @@ public sealed class QueryStoreWideFleetEligibilityLiveTests $"SELECT drop_chunks('collect.query_store_stats', older_than => TIMESTAMP '{S.AddDays(1):yyyy-MM-dd HH:mm:ss}')", connection); await drop.ExecuteNonQueryAsync(ct); return (scratch, postgres, connection); + } + catch + { + if (postgres is not null) + { + await postgres.DisposeAsync(); + } + + if (connection is not null) + { + await connection.DisposeAsync(); + } + + await scratch.DisposeAsync(); + throw; + } } - private static async Task EndAsync(NpgsqlConnection connection, int id, CancellationToken ct) + private static async Task EndAsync(NpgsqlConnection connection, CancellationToken ct) { - await using var cmd = new NpgsqlCommand($"SELECT MAX(applied_through) FROM collect.query_store_interval_wide_coverage", connection); + await using var cmd = new NpgsqlCommand("SELECT MAX(applied_through) FROM collect.query_store_interval_wide_coverage", connection); return (DateTime)(await cmd.ExecuteScalarAsync(ct))!; } @@ -125,48 +144,68 @@ public async Task FleetCheck_ReadsTheStoreWideFloorsOnce_AndEqualsTheSerialAnswe Assert.SkipWhen(string.IsNullOrEmpty(baseCs), "Set DARLING_TEST_PG to a Postgres connection string to run the #4605 live test."); var ct = TestContext.Current.CancellationToken; var ids = new[] { -4605101, -4605102, -4605103, -4605104, -4605105, -4605106 }; - var (scratch, _, connection) = await SeedAsync(baseCs!, ids, null, ct); + var (scratch, postgres, connection) = await SeedAsync(baseCs!, ids, null, ct); await using var _s = scratch; await using var _c = connection; - var end = await EndAsync(connection, ids[0], ct); + await using var _p = postgres; + var end = await EndAsync(connection, ct); var expected = await SerialAsync(connection, end, ct); Assert.True(expected.Eligible); Assert.Equal(S.AddHours(14), expected.WideStart); Assert.Equal("qsiw-fleet-03", expected.SettingServer); - var floorReads = 0; - var scratchDb = new NpgsqlConnectionStringBuilder(scratch.ConnectionString).Database; - using var listener = new ActivityListener + /* Scoped to this scratch database by name, so a class running in parallel against another database is + invisible to the count. */ + using var floorReads = new NpgsqlCommandCounter(scratch.DatabaseName, "table_is_hypertable", "raw_floor"); + using var controlReads = new NpgsqlCommandCounter(scratch.DatabaseName, "fleet_floor_count_control"); + await using (var control = postgres.CreateCommand("SELECT 1 AS fleet_floor_count_control")) { - ShouldListenTo = src => src.Name == "Npgsql", - Sample = (ref ActivityCreationOptions _) => ActivitySamplingResult.AllDataAndRecorded, - ActivityStopped = a => - { - if (!string.Equals(a.GetTagItem("db.namespace") as string, scratchDb, StringComparison.Ordinal)) - { - return; - } - - var text = (a.GetTagItem("db.query.text") ?? a.GetTagItem("db.statement")) as string; - if (text is not null && text.Contains("table_is_hypertable", StringComparison.Ordinal) && text.Contains("raw_floor", StringComparison.Ordinal)) - { - Interlocked.Increment(ref floorReads); - } - }, - }; - ActivitySource.AddActivityListener(listener); - - await using var postgres = NpgsqlDataSource.Create(scratch.ConnectionString); + await control.ExecuteScalarAsync(ct); + } + + Assert.True(controlReads.Count == 1, + $"The Npgsql activity listener counted {controlReads.Count} control commands on '{scratch.DatabaseName}' instead of 1, so its floors count would be meaningless."); + for (var run = 0; run < 20; run++) { - floorReads = 0; + var before = floorReads.Count; var got = await DarlingWebEndpoints.ResolveQueryStoreWideEligibleAsync(postgres, null, S, end, end, ct); Assert.Equal(expected, got); - Assert.Equal(1, floorReads); + Assert.Equal(1, floorReads.Count - before); } } + [Fact] + public async Task FleetCheck_WhenTheFirstServerRefusesBeforeTheFloors_ReadsNoFloors_AndAnswersRaw() + { + var baseCs = Environment.GetEnvironmentVariable("DARLING_TEST_PG"); + Assert.SkipWhen(string.IsNullOrEmpty(baseCs), "Set DARLING_TEST_PG to a Postgres connection string to run the #4605 live test."); + var ct = TestContext.Current.CancellationToken; + var ids = new[] { -4605301, -4605302, -4605303 }; + var (scratch, postgres, connection) = await SeedAsync(baseCs!, ids, null, ct); + await using var _s = scratch; + await using var _c = connection; + await using var _p = postgres; + var end = await EndAsync(connection, ct); + + /* A literal end before applied_through refuses at the step ahead of the floors read. */ + var literalEnd = S.AddHours(13); + using var floorReads = new NpgsqlCommandCounter(scratch.DatabaseName, "table_is_hypertable", "raw_floor"); + using var controlReads = new NpgsqlCommandCounter(scratch.DatabaseName, "fleet_floor_count_control"); + await using (var control = postgres.CreateCommand("SELECT 1 AS fleet_floor_count_control")) + { + await control.ExecuteScalarAsync(ct); + } + + Assert.True(controlReads.Count == 1, + $"The Npgsql activity listener counted {controlReads.Count} control commands on '{scratch.DatabaseName}' instead of 1, so its floors count would be meaningless."); + + var got = await DarlingWebEndpoints.ResolveQueryStoreWideEligibleAsync(postgres, null, S, end, literalEnd, ct); + Assert.False(got.Eligible); + Assert.Equal(0, floorReads.Count); + } + [Fact] public async Task FleetCheck_WithOneServerHoldingLegacyRows_ReadsRaw_LikeTheSerialAnswer() { @@ -178,7 +217,7 @@ public async Task FleetCheck_WithOneServerHoldingLegacyRows_ReadsRaw_LikeTheSeri await using var _s = scratch; await using var _c = connection; await using var _p = postgres; - var end = await EndAsync(connection, ids[0], ct); + var end = await EndAsync(connection, ct); Assert.Equal(default, await SerialAsync(connection, end, ct)); var got = await DarlingWebEndpoints.ResolveQueryStoreWideEligibleAsync(postgres, null, S, end, end, ct); diff --git a/Darling/PerformanceMonitor.Darling.Service/DarlingWebEndpoints.cs b/Darling/PerformanceMonitor.Darling.Service/DarlingWebEndpoints.cs index 8396d91884..588fb08ca5 100644 --- a/Darling/PerformanceMonitor.Darling.Service/DarlingWebEndpoints.cs +++ b/Darling/PerformanceMonitor.Darling.Service/DarlingWebEndpoints.cs @@ -1539,8 +1539,9 @@ so a test can run it without a live Postgres round trip. */ var bound = QueryStoreIntervalWide.WideStartBound.Window; string? settingServer = null; - /* The floors are one store-wide answer: read once here, not once per server. */ - var storeWide = await QueryStoreIntervalWide.ReadStoreWideInputsAsync(connection, McpCommandDeadlines.ReadSeconds, cancellationToken); + /* The floors are one store-wide answer: the first server to reach that step reads them into this + cache and the rest reuse them; a check that refuses before that step never reads them. */ + var storeWide = new QueryStoreIntervalWide.StoreWideInputsCache(); foreach (var (serverId, serverName) in wideServers) { var plan = await QueryStoreIntervalWide.ResolveReadAsync( diff --git a/Darling/PerformanceMonitor.Darling.Storage/QueryStoreIntervalWide.cs b/Darling/PerformanceMonitor.Darling.Storage/QueryStoreIntervalWide.cs index 3bc4d61480..4125e4f4f0 100644 --- a/Darling/PerformanceMonitor.Darling.Storage/QueryStoreIntervalWide.cs +++ b/Darling/PerformanceMonitor.Darling.Storage/QueryStoreIntervalWide.cs @@ -978,33 +978,18 @@ public static (DateTime Start, WideStartBound Bound)? ExactBelowFloorStart( /// , which takes no parameter and reads TimescaleDB's catalog for the whole store. /// ('s coverage and pending columns, , /// and the cadence probe all filter on the server, so they stay per server.) - /// A caller that resolves many servers for one window reads this once and passes it to each call. + /// A caller that resolves many servers for one window passes one to each call. /// public sealed record StoreWideInputs(DateTime? RawFloor, bool TableIsHypertable, DateTime? TableFloor); - /// Reads once. Null when TimescaleDB's catalog is absent, which - /// is the answer would reach itself (it never reads the floors then). - public static async Task ReadStoreWideInputsAsync( - NpgsqlConnection connection, int commandTimeoutSeconds, CancellationToken cancellationToken) + /// + /// A per-check holder for . fills it the first + /// time a server reaches the floors step and reuses it for every later server, so the floors are read at + /// most once per check and not at all when every server refuses before that step. + /// + public sealed class StoreWideInputsCache { - bool hasTimescale; - await using (var probe = new NpgsqlCommand("SELECT to_regclass('timescaledb_information.chunks') IS NOT NULL;", connection) { CommandTimeout = commandTimeoutSeconds }) - { - hasTimescale = (bool)(await probe.ExecuteScalarAsync(cancellationToken))!; - } - - if (!hasTimescale) - { - return null; - } - - await using var floors = new NpgsqlCommand(ChunkFloorsSql, connection) { CommandTimeout = commandTimeoutSeconds }; - await using var reader = await floors.ExecuteReaderAsync(cancellationToken); - await reader.ReadAsync(cancellationToken); - return new StoreWideInputs( - reader.IsDBNull(0) ? null : reader.GetDateTime(0), - reader.GetBoolean(1), - reader.IsDBNull(2) ? null : reader.GetDateTime(2)); + internal StoreWideInputs? Value; } /// @@ -1023,7 +1008,7 @@ public static async Task ResolveReadAsync( int commandTimeoutSeconds, ILogger? logger, CancellationToken cancellationToken, - StoreWideInputs? storeWide = null) + StoreWideInputsCache? storeWide = null) { /* The probes bind these as timestamp-without-time-zone, which Npgsql refuses for Kind=Utc; a caller that resolves its window from DateTime.UtcNow (the MCP tool) would otherwise fail every decision and read raw. */ @@ -1064,11 +1049,11 @@ remaining round trips (ChunkFloorsSql's metadata read and, worse, PlainTableFloo DateTime? rawFloor = null; DateTime? tableFloor = null; var tableIsHypertable = false; - if (hasTimescale && storeWide is not null) + if (hasTimescale && storeWide?.Value is StoreWideInputs cached) { - rawFloor = storeWide.RawFloor; - tableIsHypertable = storeWide.TableIsHypertable; - tableFloor = storeWide.TableFloor; + rawFloor = cached.RawFloor; + tableIsHypertable = cached.TableIsHypertable; + tableFloor = cached.TableFloor; } else if (hasTimescale) { @@ -1078,6 +1063,10 @@ remaining round trips (ChunkFloorsSql's metadata read and, worse, PlainTableFloo rawFloor = reader.IsDBNull(0) ? null : reader.GetDateTime(0); tableIsHypertable = reader.GetBoolean(1); tableFloor = reader.IsDBNull(2) ? null : reader.GetDateTime(2); + if (storeWide is not null) + { + storeWide.Value = new StoreWideInputs(rawFloor, tableIsHypertable, tableFloor); + } } /* Clause 2 (filledSince <= max(rawFloor, windowStart)) needs only rawFloor, already in hand from From e431fafb5f20e5ad8cb4e46abe065a798ec70d54 Mon Sep 17 00:00:00 2001 From: Erik Darling <2136037+erikdarlingdata@users.noreply.github.com> Date: Sat, 3 Oct 2026 01:29:13 -0400 Subject: [PATCH 5/5] Use synthetic server names in the fleet eligibility test (#4605) --- .../QueryStoreWideFleetEligibilityLiveTests.cs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/Darling/Darling.Tests/QueryStoreWideFleetEligibilityLiveTests.cs b/Darling/Darling.Tests/QueryStoreWideFleetEligibilityLiveTests.cs index 209e8f6f18..31e5be41a6 100644 --- a/Darling/Darling.Tests/QueryStoreWideFleetEligibilityLiveTests.cs +++ b/Darling/Darling.Tests/QueryStoreWideFleetEligibilityLiveTests.cs @@ -59,12 +59,12 @@ public sealed class QueryStoreWideFleetEligibilityLiveTests await using var ins = new NpgsqlCommand( "INSERT INTO servers (server_id, server_name, display_name, is_enabled, sql_major_version, created_date, modified_date) " - + $"VALUES ({id}, 'qsiw-fleet-{i:00}', 'qsiw-fleet-{i:00}', TRUE, 16, now(), now()) ON CONFLICT (server_id) DO UPDATE SET is_enabled = TRUE", connection); + + $"VALUES ({id}, 'qsiw-fleetsrv-{(char)('a' + i)}', 'qsiw-fleetsrv-{(char)('a' + i)}', TRUE, 16, now(), now()) ON CONFLICT (server_id) DO UPDATE SET is_enabled = TRUE", connection); await ins.ExecuteNonQueryAsync(ct); await QueryStoreIntervalWideBelowFloorLiveTests.SeedQueryStoreLogAsync(connection, id, S.AddDays(-60), S.AddDays(4), null, null, ct); } - /* Different read starts: servers 0 and 3 share the latest claim, so the tie goes to the lower server_id, qsiw-fleet-03. */ + /* Different read starts: servers 0 and 3 share the latest claim, so the tie goes to the lower server_id, qsiw-fleetsrv-d. */ for (var i = 0; i < eligibleIds.Length; i++) { var hours = i == 0 || i == 3 ? 14 : 12 + (i % 2); @@ -153,7 +153,7 @@ public async Task FleetCheck_ReadsTheStoreWideFloorsOnce_AndEqualsTheSerialAnswe var expected = await SerialAsync(connection, end, ct); Assert.True(expected.Eligible); Assert.Equal(S.AddHours(14), expected.WideStart); - Assert.Equal("qsiw-fleet-03", expected.SettingServer); + Assert.Equal("qsiw-fleetsrv-d", expected.SettingServer); /* Scoped to this scratch database by name, so a class running in parallel against another database is invisible to the count. */