diff --git a/Darling/Darling.Tests/AvailabilityGroupCountReadTests.cs b/Darling/Darling.Tests/AvailabilityGroupCountReadTests.cs index 42ac575fca..2a1d411bf5 100644 --- a/Darling/Darling.Tests/AvailabilityGroupCountReadTests.cs +++ b/Darling/Darling.Tests/AvailabilityGroupCountReadTests.cs @@ -31,7 +31,7 @@ public sealed class AvailabilityGroupCountReadTests private static string? ConnectionString => Environment.GetEnvironmentVariable("DARLING_TEST_PG"); /// Plants TWO distinct AGs under one server, so a count/topology pair that "agree" by both - /// landing on 0 or both landing on 1 does not pass this — 's + /// landing on 0 or both landing on 1 does not pass this — 's /// doc explains why the database grain and the ORDER BY the topology read carries cannot change the /// number either read reports. [Fact] diff --git a/Darling/Darling.Tests/DarlingAgReaderTests.cs b/Darling/Darling.Tests/DarlingAgReaderTests.cs index 9c1fa4b471..2211b86670 100644 --- a/Darling/Darling.Tests/DarlingAgReaderTests.cs +++ b/Darling/Darling.Tests/DarlingAgReaderTests.cs @@ -392,98 +392,8 @@ public void SerializedShape_CarriesTheFieldsBothConsumersRead() JsonAssert.DoesNotContain("\"severity\": 3", json); } - /* ─────────────────────────── SQL dialect pins ─────────────────────────── */ - - [Fact] - public void ReadSql_IsPostgresDialect_WithPositionalServerFilter() - { - foreach (var sql in new[] { Reader.ReplicaStatesSql, Reader.DatabaseReplicaStatesSql }) - { - /* Positional params (Npgsql), never a named @param or an N'' literal. */ - Assert.Contains("$1", sql, StringComparison.Ordinal); - Assert.DoesNotContain("@", sql, StringComparison.Ordinal); - Assert.DoesNotContain("N'", sql, StringComparison.Ordinal); - - /* The filter guards BOTH the inner latest-instant aggregate and the outer read — count, don't just - Contains, or deleting either WHERE leaves this green. */ - Assert.Equal(2, Occurrences(sql, "$1::integer IS NULL")); - } - } - - private static int Occurrences(string haystack, string needle) - { - var count = 0; - for (var i = haystack.IndexOf(needle, StringComparison.Ordinal); i >= 0; - i = haystack.IndexOf(needle, i + needle.Length, StringComparison.Ordinal)) - { - count++; - } - - return count; - } - - [Fact] - public void ReadSql_RestrictsToTheEnabledServerRegistry() - { - /* Every other fleet read joins the enabled registry; without it a server disabled in the control plane - keeps producing AG cards until its rows age out, after it has already left /api/fleet and the sidebar. */ - foreach (var sql in new[] { Reader.ReplicaStatesSql, Reader.DatabaseReplicaStatesSql }) - { - Assert.Contains("JOIN servers AS s", sql, StringComparison.Ordinal); - Assert.Contains("s.is_enabled", sql, StringComparison.Ordinal); - } - } - - [Fact] - public void ReadSql_KeepsEveryRowAtEachServersNewestCollection() - { - /* DISTINCT ON would keep ONE row per server — which is exactly wrong here, since a snapshot is many rows - (one per replica / per database). The join against MAX(collection_time) keeps them all. */ - foreach (var sql in new[] { Reader.ReplicaStatesSql, Reader.DatabaseReplicaStatesSql }) - { - Assert.Contains("MAX(collection_time)", sql, StringComparison.Ordinal); - Assert.Contains("GROUP BY server_id", sql, StringComparison.Ordinal); - Assert.DoesNotContain("DISTINCT ON", sql, StringComparison.Ordinal); - } - } - - [Fact] - public void ReadSql_SelectsIsLocalOnBothGrains() - { - /* V37 put is_local on the replica grain; V34 already had it on the database grain. Inserting it shifted - the replica read's ordinals, which is the classic silent mis-map. */ - Assert.Contains("r.is_local", Reader.ReplicaStatesSql, StringComparison.Ordinal); - Assert.Contains("d.is_local", Reader.DatabaseReplicaStatesSql, StringComparison.Ordinal); - } - - [Fact] - public void ReadSql_ReadsTheCollectQualifiedBaseTables() - { - /* Post-V14 collectors get no v_* passthrough view, so the readers name the base tables directly. */ - Assert.Contains("FROM collect.ag_replica_states", Reader.ReplicaStatesSql, StringComparison.Ordinal); - Assert.Contains("FROM collect.ag_database_replica_states", Reader.DatabaseReplicaStatesSql, StringComparison.Ordinal); - } - - [Fact] - public void ReadSql_SelectsEveryColumnTheDtoSurfaces() - { - foreach (var column in new[] - { - "role_desc", "operational_state_desc", "connected_state_desc", "recovery_health_desc", - "synchronization_health_desc", "availability_mode_desc", "failover_mode_desc", "endpoint_url", - }) - { - Assert.Contains(column, Reader.ReplicaStatesSql, StringComparison.Ordinal); - } - - foreach (var column in new[] - { - "synchronization_state_desc", "last_hardened_lsn", "last_commit_lsn", "log_send_queue_size", - "redo_queue_size", "log_send_rate", "redo_rate", "is_suspended", "suspend_reason_desc", - "secondary_lag_seconds", "is_local", - }) - { - Assert.Contains(column, Reader.DatabaseReplicaStatesSql, StringComparison.Ordinal); - } - } + /* The SQL dialect/shape pins used to live here, against this reader's own copy of the statement text. That + copy moved to DarlingAgStatesReaderTests (#4228) along with the statement text itself — see + DarlingAgStatesReader in PerformanceMonitor.Darling.Storage, now the one implementation ReadReplicasAsync + and ReadDatabasesAsync above map into this file's ReplicaRow / DatabaseRow. */ } diff --git a/Darling/Darling.Tests/DarlingAgStatesReaderLiveEqualityTests.cs b/Darling/Darling.Tests/DarlingAgStatesReaderLiveEqualityTests.cs new file mode 100644 index 0000000000..93adadba9a --- /dev/null +++ b/Darling/Darling.Tests/DarlingAgStatesReaderLiveEqualityTests.cs @@ -0,0 +1,597 @@ +/* + * 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.Generic; +using System.Globalization; +using System.Linq; +using System.Text; +using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; +using Npgsql; +using NpgsqlTypes; +using PerformanceMonitor.Collectors; +using PerformanceMonitor.Common; +using PerformanceMonitor.Darling.Service.Mcp; +using PerformanceMonitor.Darling.Storage; +using PerformanceMonitor.Darling.Viewer; +using Xunit; + +namespace Darling.Tests; + +/// +/// #4228: the live proof beside 's string-level pins. This class seeds +/// ten server shapes (current, stopped 2 days ago, stopped 2 hours ago, disabled-with-current-rows, and a +/// server removed from config_monitored_servers but still present in servers) across several +/// daily chunks, then proves two things against a real store: (1) every entry point the move touches — the +/// viewer's AG tab, /api/ag, /api/ag/count and get_ag_health, with and without a server +/// filter — returns exactly the rows the pre-#4228 unbounded statements returned; and (2) the floor-bounded +/// step two plans meaningfully fewer chunks than the unbounded oracle for both the shared-floor batch and a +/// stale server's own point read. +/// +[Collection("live-postgres")] +public sealed class DarlingAgStatesReaderLiveEqualityTests +{ + private const string AgName = "AG4228"; + + private static readonly string[] CurrentNames = + { + "ag4228-cur-1", "ag4228-cur-2", "ag4228-cur-3", "ag4228-cur-4", "ag4228-cur-5", "ag4228-cur-6", + }; + + private const string StaleName = "ag4228-stale"; + private const string InsideHorizonName = "ag4228-inside-horizon"; + private const string DisabledName = "ag4228-disabled"; + private const string RemovedName = "ag4228-removed"; + + private static readonly string[] AllNames = CurrentNames + .Concat(new[] { StaleName, InsideHorizonName, DisabledName, RemovedName }) + .ToArray(); + + private static int Id(string name) => ServerIdHelper.GetDeterministicHashCode(name); + + /* ── the pre-#4228 shapes, kept here as the row-correctness oracle (bare table names — equivalent to the + Service's shipped collect.-qualified text under this store's search_path, like DarlingAgStatesReader's + own doc explains) ── */ + + private const string OldReplicaStatesSql = """ + SELECT + r.server_id, r.server_name, r.collection_time, r.ag_name, r.replica_server_name, r.role_desc, + r.is_local, r.operational_state_desc, r.connected_state_desc, r.recovery_health_desc, + r.synchronization_health_desc, r.availability_mode_desc, r.failover_mode_desc, r.endpoint_url + FROM ag_replica_states AS r + JOIN + ( + SELECT server_id, MAX(collection_time) AS max_collection_time + FROM ag_replica_states + WHERE ($1::integer IS NULL OR server_id = $1) + GROUP BY server_id + ) AS latest + ON r.server_id = latest.server_id + AND r.collection_time = latest.max_collection_time + JOIN servers AS s + ON s.server_id = r.server_id + AND s.is_enabled + WHERE ($1::integer IS NULL OR r.server_id = $1) + ORDER BY r.server_name, r.ag_name, r.replica_server_name + """; + + private const string OldDatabaseReplicaStatesSql = """ + SELECT + d.server_id, d.server_name, d.collection_time, d.ag_name, d.database_name, d.replica_server_name, + d.is_local, d.synchronization_state_desc, d.last_hardened_lsn, d.last_commit_lsn, + d.log_send_queue_size, d.redo_queue_size, d.log_send_rate, d.redo_rate, d.is_suspended, + d.suspend_reason_desc, d.availability_mode_desc, d.secondary_lag_seconds + FROM ag_database_replica_states AS d + JOIN + ( + SELECT server_id, MAX(collection_time) AS max_collection_time + FROM ag_database_replica_states + WHERE ($1::integer IS NULL OR server_id = $1) + GROUP BY server_id + ) AS latest + ON d.server_id = latest.server_id + AND d.collection_time = latest.max_collection_time + JOIN servers AS s + ON s.server_id = d.server_id + AND s.is_enabled + WHERE ($1::integer IS NULL OR d.server_id = $1) + ORDER BY d.server_name, d.ag_name, d.database_name, d.replica_server_name + """; + + private const string OldAvailabilityGroupCountSql = """ + SELECT COUNT(DISTINCT (r.server_id, UPPER(COALESCE(r.ag_name, '')))) + FROM ag_replica_states AS r + JOIN + ( + SELECT server_id, MAX(collection_time) AS max_collection_time + FROM ag_replica_states + WHERE ($1::integer IS NULL OR server_id = $1) + GROUP BY server_id + ) AS latest + ON r.server_id = latest.server_id + AND r.collection_time = latest.max_collection_time + JOIN servers AS s + ON s.server_id = r.server_id + AND s.is_enabled + WHERE ($1::integer IS NULL OR r.server_id = $1) + """; + + /* ── seed statements: $1 collection_id, $2 collection_time, $3 server_id, $4 server_name, $5 ag_name ── */ + + private const string ReplicaNode1Sql = """ + INSERT INTO ag_replica_states + (collection_id, collection_time, server_id, server_name, ag_name, replica_server_name, role_desc, + operational_state_desc, connected_state_desc, recovery_health_desc, synchronization_health_desc, + availability_mode_desc, failover_mode_desc, endpoint_url) + VALUES ($1, $2, $3, $4, $5, 'NODE1', 'PRIMARY', 'ONLINE', 'CONNECTED', 'ONLINE', 'HEALTHY', + 'SYNCHRONOUS_COMMIT', 'AUTOMATIC', 'TCP://NODE1:5022') + """; + + private const string ReplicaNode2Sql = """ + INSERT INTO ag_replica_states + (collection_id, collection_time, server_id, server_name, ag_name, replica_server_name, role_desc, + operational_state_desc, connected_state_desc, recovery_health_desc, synchronization_health_desc, + availability_mode_desc, failover_mode_desc, endpoint_url) + VALUES ($1, $2, $3, $4, $5, 'NODE2', 'SECONDARY', NULL, 'CONNECTED', 'ONLINE', 'HEALTHY', + 'SYNCHRONOUS_COMMIT', 'AUTOMATIC', 'TCP://NODE2:5022') + """; + + private const string DatabaseDb1Sql = """ + INSERT INTO ag_database_replica_states + (collection_id, collection_time, server_id, server_name, ag_name, database_name, replica_server_name, + is_local, synchronization_state_desc, last_hardened_lsn, last_commit_lsn, log_send_queue_size, + redo_queue_size, log_send_rate, redo_rate, is_suspended, suspend_reason_desc, availability_mode_desc, + secondary_lag_seconds) + VALUES ($1, $2, $3, $4, $5, 'DB1', 'NODE2', false, 'SYNCHRONIZED', '0x01', '0x02', 500, 0, 100, 0, false, + NULL, 'SYNCHRONOUS_COMMIT', 1) + """; + + private const string DatabaseDb2Sql = """ + INSERT INTO ag_database_replica_states + (collection_id, collection_time, server_id, server_name, ag_name, database_name, replica_server_name, + is_local, synchronization_state_desc, last_hardened_lsn, last_commit_lsn, log_send_queue_size, + redo_queue_size, log_send_rate, redo_rate, is_suspended, suspend_reason_desc, availability_mode_desc, + secondary_lag_seconds) + VALUES ($1, $2, $3, $4, $5, 'DB2', 'NODE2', false, 'SYNCHRONIZED', '0x03', '0x04', 200, 0, 100, 0, false, + NULL, 'SYNCHRONOUS_COMMIT', 2) + """; + + [Fact] + public async Task AgStatesReader_MatchesTheOldUnboundedStatements_AcrossEveryServerShape_AgainstDevPostgres() + { + var cs = Environment.GetEnvironmentVariable("DARLING_TEST_PG"); + Assert.SkipWhen(string.IsNullOrEmpty(cs), "Set DARLING_TEST_PG to a Postgres connection string to run the live #4228 AG read-equality test."); + + var ct = TestContext.Current.CancellationToken; + using var connection = new NpgsqlConnection(cs); + await connection.OpenAsync(ct); + await PgMigrations.MigrateAsync(connection, ct); + await DeleteSentinelRowsAsync(connection, ct); + await using var postgres = NpgsqlDataSource.Create(cs!); + + var bodySucceeded = false; + try + { + var now = Micro(DateTime.UtcNow); + await SeedAsync(connection, now, ct); + + var sentinelIds = new HashSet(AllNames.Select(Id)); + var disabledId = Id(DisabledName); + var staleId = Id(StaleName); + var insideHorizonId = Id(InsideHorizonName); + var removedId = Id(RemovedName); + var cur1Id = Id(CurrentNames[0]); + + /* Unfiltered, scoped to this test's own servers: one equality assertion over all nine visible + shapes' rows proves, in one shot, that disabled exclusion, stale/inside-horizon/removed + inclusion at their true instants, and the tie at each server's newest snapshot all still match + what the unbounded oracle returns — a wrong row anywhere breaks set equality. */ + var oldReplica = Scoped(await ReadOracleRowsAsync(connection, OldReplicaStatesSql, null, ct), sentinelIds); + var newReplicaTyped = await DarlingAgStatesReader.GetReplicaStatesAsync(postgres, null, now, ct); + var newReplica = newReplicaTyped.Where(r => sentinelIds.Contains(r.ServerId)) + .Select(RenderReplica).OrderBy(s => s, StringComparer.Ordinal).ToList(); + Assert.Equal(oldReplica, newReplica); + Assert.Equal(18, newReplica.Count); + + var oldDatabase = Scoped(await ReadOracleRowsAsync(connection, OldDatabaseReplicaStatesSql, null, ct), sentinelIds); + var newDatabaseTyped = await DarlingAgStatesReader.GetDatabaseReplicaStatesAsync(postgres, null, now, ct); + var newDatabase = newDatabaseTyped.Where(r => sentinelIds.Contains(r.ServerId)) + .Select(RenderDatabase).OrderBy(s => s, StringComparer.Ordinal).ToList(); + Assert.Equal(oldDatabase, newDatabase); + Assert.Equal(18, newDatabase.Count); + + /* The tie: both replicas of cur-1's newest snapshot share the identical collection_time — the + normal shape of every AG snapshot, not an edge case. collection_time is NOT NULL on both AG + tables (schema-enforced, V34), so a tie is the only ordering edge case reachable here. */ + var cur1Replicas = newReplicaTyped.Where(r => r.ServerId == cur1Id).ToList(); + Assert.Equal(2, cur1Replicas.Count); + Assert.Equal(cur1Replicas[0].CollectionTime, cur1Replicas[1].CollectionTime); + Assert.Equal(now, cur1Replicas[0].CollectionTime); + + /* Stale: its own read returns its own true newest instant (2 days old), not the shared floor's. */ + var staleReplicas = newReplicaTyped.Where(r => r.ServerId == staleId).ToList(); + Assert.Equal(2, staleReplicas.Count); + Assert.Equal(now.AddDays(-2), staleReplicas[0].CollectionTime); + + /* The serverIdFilter path end to end, for the shapes most likely to break it. */ + foreach (var (_, id) in new[] { (StaleName, staleId), (InsideHorizonName, insideHorizonId), (RemovedName, removedId) }) + { + var oldRows = (await ReadOracleRowsAsync(connection, OldReplicaStatesSql, id, ct)) + .Select(r => r.Rendered).OrderBy(s => s, StringComparer.Ordinal).ToList(); + var newRows = (await DarlingAgStatesReader.GetReplicaStatesAsync(postgres, id, now, ct)) + .Select(RenderReplica).OrderBy(s => s, StringComparer.Ordinal).ToList(); + Assert.Equal(oldRows, newRows); + Assert.Equal(2, newRows.Count); + } + + /* Disabled, filtered directly to it: zero rows even though its rows are the newest in the table — + the output-side is_enabled join excludes it exactly as today. */ + var disabledOld = await ReadOracleRowsAsync(connection, OldReplicaStatesSql, disabledId, ct); + var disabledNew = await DarlingAgStatesReader.GetReplicaStatesAsync(postgres, disabledId, now, ct); + Assert.Empty(disabledOld); + Assert.Empty(disabledNew); + + /* Group count (backs GetAvailabilityGroupCountAsync / /api/ag/count): one AG per non-disabled + server, zero for disabled. */ + foreach (var name in AllNames) + { + var id = Id(name); + var expected = name == DisabledName ? 0 : 1; + Assert.Equal(expected, await ReadOracleScalarAsync(connection, OldAvailabilityGroupCountSql, id, ct)); + Assert.Equal(expected, await DarlingAgStatesReader.GetReplicaGroupCountAsync(postgres, id, now, ct)); + } + + /* Entry points: the viewer read, /api/ag, /api/ag/count and get_ag_health, with and without a + server filter — wiring coverage on top of the row-level proof above, not a second oracle. */ + await using (var viewer = new ViewerDataService(cs!)) + { + var cards = (await viewer.GetAvailabilityGroupsAsync(ct)).Where(c => sentinelIds.Contains(c.ServerId)).ToList(); + Assert.Equal(9, cards.Count); + Assert.DoesNotContain(cards, c => c.ServerId == disabledId); + var staleCard = Assert.Single(cards, c => c.ServerId == staleId); + Assert.Equal(now.AddDays(-2), staleCard.CollectionTime); + Assert.Equal(2, staleCard.Replicas.Count); + var removedCard = Assert.Single(cards, c => c.ServerId == removedId); + Assert.Equal(2, removedCard.Databases.Count); + } + + var fleetHealth = await DarlingAgReader.GetAgHealthAsync(postgres, null, now, ct); + var fleetGroups = fleetHealth.AvailabilityGroups.Where(g => sentinelIds.Contains(g.ServerId)).ToList(); + Assert.Equal(9, fleetGroups.Count); + Assert.DoesNotContain(fleetGroups, g => g.ServerId == disabledId); + + var fleetCount = await DarlingAgReader.GetAvailabilityGroupCountAsync(postgres, null, ct); + Assert.Equal(fleetHealth.AvailabilityGroupCount, fleetCount); + + var scopedHealth = await DarlingAgReader.GetAgHealthAsync(postgres, staleId, now, ct); + Assert.Equal(StaleName, Assert.Single(scopedHealth.AvailabilityGroups).ServerName); + + var fleetJson = await DarlingMcpAgTools.GetAgHealth(postgres, null); + Assert.False(McpHelpers.IsErrorEnvelope(fleetJson), $"tool returned an error: {fleetJson}"); + using (var doc = JsonDocument.Parse(fleetJson)) + { + var groups = doc.RootElement.GetProperty("availability_groups").EnumerateArray() + .Where(g => sentinelIds.Contains(g.GetProperty("server_id").GetInt32())).ToList(); + Assert.Equal(9, groups.Count); + } + + var namedJson = await DarlingMcpAgTools.GetAgHealth(postgres, CurrentNames[0]); + Assert.False(McpHelpers.IsErrorEnvelope(namedJson), $"tool returned an error: {namedJson}"); + using (var doc = JsonDocument.Parse(namedJson)) + { + var group = Assert.Single(doc.RootElement.GetProperty("availability_groups").EnumerateArray()); + Assert.Equal(CurrentNames[0], group.GetProperty("server_name").GetString()); + Assert.Equal(2, group.GetProperty("replicas").GetArrayLength()); + Assert.Equal(2, group.GetProperty("databases").GetArrayLength()); + } + + bodySucceeded = true; + } + finally + { + await LiveStoreCleanup.RunAsync(cs!, bodySucceeded, DeleteSentinelRowsAsync); + } + } + + [Fact] + public async Task AgStatesReader_StepTwoStaysBounded_ForCurrentAndStaleServers_AgainstDevPostgres() + { + var cs = Environment.GetEnvironmentVariable("DARLING_TEST_PG"); + Assert.SkipWhen(string.IsNullOrEmpty(cs), "Set DARLING_TEST_PG to a Postgres connection string to run the live #4228 AG plan test."); + + var ct = TestContext.Current.CancellationToken; + using var connection = new NpgsqlConnection(cs); + await connection.OpenAsync(ct); + await PgMigrations.MigrateAsync(connection, ct); + var timescale = await PrepareHypertablesAsync(cs!, connection, ct); + Assert.SkipUnless(timescale, "TimescaleDB is not available on this store: there are no chunks to exclude."); + + await DeleteSentinelRowsAsync(connection, ct); + var bodySucceeded = false; + try + { + var now = Micro(DateTime.UtcNow); + await SeedAsync(connection, now, ct); + var sentinelIds = new HashSet(AllNames.Select(Id)); + + /* The real step-one + split, exactly as GetReplicaStatesAsync runs it (scoped to this test's own + servers so the floor used below is not perturbed by whatever else the shared store holds). */ + var instants = (await ReadInstantsAsync(connection, DarlingAgStatesReader.ReplicaNewestInstantSql, ct)) + .Where(i => sentinelIds.Contains(i.ServerId)).ToList(); + var (floor, stale) = DarlingAgStatesReader.SplitCurrentAndStale(instants, now); + Assert.NotNull(floor); + var staleServer = stale.Single(s => s.ServerId == Id(StaleName)); + + /* Shared floor vs the unbounded oracle. Relative reduction, not an absolute ceiling: darlingtest + is shared with the rest of the suite, and an unrelated class's own near-"now" seed can add a + chunk this seed did not create (FleetReadsAreBoundedByTheFleetTests and + EventWindowedReadsAreBoundedLivePostgresTests hit the same thing first, same fix). Confirmed + once by hand against a freshly created database — see the PR body's measured table — that the + floored plan touches at most 2 chunks for this exact seed when every server is current. */ + var oldPlan = await ExplainAsync(connection, "EXPLAIN (COSTS OFF) " + OldReplicaStatesSql, new[] { ServerFilterParam(null) }, ct); + var newPlan = await ExplainAsync(connection, "EXPLAIN (COSTS OFF) " + DarlingAgStatesReader.ReplicaStatesSql, + new[] { TimestampParam(floor!.Value), ServerFilterParam(null) }, ct); + var oldChunks = PlanChunkScans.Count(oldPlan); + var newChunks = PlanChunkScans.Count(newPlan); + Assert.True(oldChunks - newChunks >= 3, + $"expected the floor to exclude several of the seeded old-day chunks (old={oldChunks}, new={newChunks}):\nOLD:\n{oldPlan}\nNEW:\n{newPlan}"); + + /* The stale server's own point read is bounded BELOW by its own newest instant, so it cannot scan + older than that — cheaper than the unbounded oracle's whole-table scan, but not necessarily a + single chunk: the WHERE clause has no upper bound, and the shared table keeps gaining new + chunks every day regardless of this one server's staleness, so a chunk that opened after this + server's own floor cannot be excluded from an open-ended ">=" predicate. */ + var oldStalePlan = await ExplainAsync(connection, "EXPLAIN (COSTS OFF) " + OldReplicaStatesSql, + new[] { ServerFilterParam(staleServer.ServerId) }, ct); + var newStalePlan = await ExplainAsync(connection, "EXPLAIN (COSTS OFF) " + DarlingAgStatesReader.ReplicaStatesSql, + new[] { TimestampParam(staleServer.Instant), ServerFilterParam(staleServer.ServerId) }, ct); + var oldStaleChunks = PlanChunkScans.Count(oldStalePlan); + var newStaleChunks = PlanChunkScans.Count(newStalePlan); + Assert.True(oldStaleChunks - newStaleChunks >= 2, + $"expected the stale server's own floor to exclude chunks older than its own newest instant (old={oldStaleChunks}, new={newStaleChunks}):\nOLD:\n{oldStalePlan}\nNEW:\n{newStalePlan}"); + + bodySucceeded = true; + } + finally + { + await LiveStoreCleanup.RunAsync(cs!, bodySucceeded, DeleteSentinelRowsAsync); + } + } + + /* ─────────────────────────── seeding ─────────────────────────── */ + + private static async Task SeedAsync(NpgsqlConnection connection, DateTime now, CancellationToken ct) + { + foreach (var name in CurrentNames) + { + await RegisterServerAsync(connection, Id(name), name, enabled: true, ct); + await SeedServerAsync(connection, Id(name), name, Instants(now, new[] { 4, 3, 2, 1 }, now), ct); + } + + await RegisterServerAsync(connection, Id(StaleName), StaleName, enabled: true, ct); + await SeedServerAsync(connection, Id(StaleName), StaleName, Instants(now, new[] { 5, 4, 3 }, now.AddDays(-2)), ct); + + await RegisterServerAsync(connection, Id(InsideHorizonName), InsideHorizonName, enabled: true, ct); + await SeedServerAsync(connection, Id(InsideHorizonName), InsideHorizonName, Instants(now, new[] { 4, 3, 2, 1 }, now.AddHours(-2)), ct); + + await RegisterServerAsync(connection, Id(DisabledName), DisabledName, enabled: false, ct); + await SeedServerAsync(connection, Id(DisabledName), DisabledName, Instants(now, new[] { 1 }, now), ct); + + await RegisterServerAsync(connection, Id(RemovedName), RemovedName, enabled: true, ct); + await SeedServerAsync(connection, Id(RemovedName), RemovedName, Instants(now, new[] { 1 }, now), ct); + await SimulateRemoveServerAsync(connection, Id(RemovedName), RemovedName, ct); + } + + /// Several days of history (2 rows/day, several distinct chunks) plus a "collected every few + /// minutes" recent window ending at — the server's newest snapshot. + private static List Instants(DateTime now, int[] daysBack, DateTime newest) + { + var list = new List(); + foreach (var d in daysBack) + { + var day = now.AddDays(-d); + list.Add(day.AddHours(3)); + list.Add(day.AddHours(15)); + } + + for (var minutesAgo = 20; minutesAgo >= 5; minutesAgo -= 5) + { + list.Add(newest.AddMinutes(-minutesAgo)); + } + + list.Add(newest); + return list; + } + + private static async Task SeedServerAsync(NpgsqlConnection connection, int serverId, string serverName, IReadOnlyList instants, CancellationToken ct) + { + foreach (var at in instants) + { + await ExecAsync(connection, ReplicaNode1Sql, ct, CollectionIdGenerator.Next(), at, serverId, serverName, AgName); + await ExecAsync(connection, ReplicaNode2Sql, ct, CollectionIdGenerator.Next(), at, serverId, serverName, AgName); + await ExecAsync(connection, DatabaseDb1Sql, ct, CollectionIdGenerator.Next(), at, serverId, serverName, AgName); + await ExecAsync(connection, DatabaseDb2Sql, ct, CollectionIdGenerator.Next(), at, serverId, serverName, AgName); + } + } + + private static async Task RegisterServerAsync(NpgsqlConnection connection, int serverId, string serverName, bool enabled, CancellationToken ct) + { + using var command = new NpgsqlCommand(""" + INSERT INTO servers (server_id, server_name, display_name, is_enabled, sql_major_version, created_date, modified_date) + VALUES ($1, $2, $3, $4, 15, $5, $5) + ON CONFLICT (server_id) DO UPDATE SET is_enabled = $4, sql_major_version = 15 + """, connection); + command.Parameters.AddWithValue(serverId); + command.Parameters.AddWithValue(serverName); + command.Parameters.AddWithValue(serverName); + command.Parameters.AddWithValue(enabled); + command.Parameters.AddWithValue(DateTime.SpecifyKind(DateTime.UtcNow, DateTimeKind.Unspecified)); + await command.ExecuteNonQueryAsync(ct); + } + + /// Mirrors add_servers' write, then remove_server's own DELETE FROM config_monitored_servers + /// WHERE server_id = $1 — the servers registry row and all collected history are untouched, exactly as + /// remove_server documents (#4228 PR body). + private static async Task SimulateRemoveServerAsync(NpgsqlConnection connection, int serverId, string serverName, CancellationToken ct) + { + using (var insert = new NpgsqlCommand( + "INSERT INTO config_monitored_servers (server_id, name, host) VALUES ($1, $2, $3) ON CONFLICT (server_id) DO NOTHING", connection)) + { + insert.Parameters.AddWithValue(serverId); + insert.Parameters.AddWithValue(serverName); + insert.Parameters.AddWithValue(serverName); + await insert.ExecuteNonQueryAsync(ct); + } + + using var delete = new NpgsqlCommand("DELETE FROM config_monitored_servers WHERE server_id = $1", connection); + delete.Parameters.AddWithValue(serverId); + await delete.ExecuteNonQueryAsync(ct); + } + + private static async Task ExecAsync(NpgsqlConnection connection, string sql, CancellationToken ct, params object?[] args) + { + using var command = new NpgsqlCommand(sql, connection); + foreach (var arg in args) + { + command.Parameters.AddWithValue(arg is DateTime dt ? DateTime.SpecifyKind(dt, DateTimeKind.Unspecified) : (arg ?? DBNull.Value)); + } + + await command.ExecuteNonQueryAsync(ct); + } + + /* ─────────────────────────── oracle + reader row rendering ─────────────────────────── */ + + private static string Cell(object? value) => value switch + { + null => "null", + DateTime dt => dt.ToString("O", CultureInfo.InvariantCulture), + _ => Convert.ToString(value, CultureInfo.InvariantCulture)!, + }; + + private static string RenderReplica(DarlingAgStatesReader.ReplicaStateRow r) => string.Join("|", + Cell(r.ServerId), Cell(r.ServerName), Cell(r.CollectionTime), Cell(r.AgName), Cell(r.ReplicaServerName), + Cell(r.RoleDesc), Cell(r.IsLocal), Cell(r.OperationalStateDesc), Cell(r.ConnectedStateDesc), + Cell(r.RecoveryHealthDesc), Cell(r.SynchronizationHealthDesc), Cell(r.AvailabilityModeDesc), + Cell(r.FailoverModeDesc), Cell(r.EndpointUrl)); + + private static string RenderDatabase(DarlingAgStatesReader.DatabaseReplicaStateRow r) => string.Join("|", + Cell(r.ServerId), Cell(r.ServerName), Cell(r.CollectionTime), Cell(r.AgName), Cell(r.DatabaseName), + Cell(r.ReplicaServerName), Cell(r.IsLocal), Cell(r.SynchronizationStateDesc), Cell(r.LastHardenedLsn), + Cell(r.LastCommitLsn), Cell(r.LogSendQueueSize), Cell(r.RedoQueueSize), Cell(r.LogSendRate), Cell(r.RedoRate), + Cell(r.IsSuspended), Cell(r.SuspendReasonDesc), Cell(r.AvailabilityModeDesc), Cell(r.SecondaryLagSeconds)); + + private static async Task> ReadOracleRowsAsync(NpgsqlConnection connection, string sql, int? serverIdFilter, CancellationToken ct) + { + using var command = new NpgsqlCommand(sql, connection); + command.Parameters.Add(ServerFilterParam(serverIdFilter)); + var rows = new List<(int, string)>(); + using var reader = await command.ExecuteReaderAsync(ct); + while (await reader.ReadAsync(ct)) + { + var cells = new string[reader.FieldCount]; + for (var i = 0; i < reader.FieldCount; i++) + { + cells[i] = Cell(reader.IsDBNull(i) ? null : reader.GetValue(i)); + } + + rows.Add((reader.GetInt32(0), string.Join("|", cells))); + } + + return rows; + } + + private static async Task ReadOracleScalarAsync(NpgsqlConnection connection, string sql, int? serverIdFilter, CancellationToken ct) + { + using var command = new NpgsqlCommand(sql, connection); + command.Parameters.Add(ServerFilterParam(serverIdFilter)); + var result = await command.ExecuteScalarAsync(ct); + return Convert.ToInt32(result ?? 0L); + } + + private static async Task> ReadInstantsAsync(NpgsqlConnection connection, string sql, CancellationToken ct) + { + using var command = new NpgsqlCommand(sql, connection); + command.Parameters.Add(ServerFilterParam(null)); + var rows = new List<(int, DateTime)>(); + using var reader = await command.ExecuteReaderAsync(ct); + while (await reader.ReadAsync(ct)) + { + rows.Add((reader.GetInt32(0), reader.GetDateTime(1))); + } + + return rows; + } + + private static List Scoped(List<(int ServerId, string Rendered)> rows, HashSet ids) => + rows.Where(r => ids.Contains(r.ServerId)).Select(r => r.Rendered).OrderBy(s => s, StringComparer.Ordinal).ToList(); + + /* ─────────────────────────── plumbing ─────────────────────────── */ + + /// Binds the optional server filter exactly as DarlingAgStatesReader.AddServerFilter does. + private static NpgsqlParameter ServerFilterParam(int? value) => new NpgsqlParameter + { + NpgsqlDbType = NpgsqlDbType.Integer, + Value = (object?)value ?? DBNull.Value, + }; + + /// Binds a floor exactly as DarlingAgStatesReader.AddTimestamp does. + private static NpgsqlParameter TimestampParam(DateTime value) => + new NpgsqlParameter { TypedValue = DateTime.SpecifyKind(value, DateTimeKind.Unspecified) }; + + private static async Task ExplainAsync(NpgsqlConnection connection, string sql, NpgsqlParameter[] args, CancellationToken ct) + { + using var command = new NpgsqlCommand(sql, connection); + command.Parameters.AddRange(args); + var plan = new StringBuilder(); + using var reader = await command.ExecuteReaderAsync(ct); + while (await reader.ReadAsync(ct)) + { + plan.AppendLine(reader.GetString(0)); + } + + return plan.ToString(); + } + + /// The store the way the service builds it — duplicated rather than shared across test files per + /// this suite's convention (see EventWindowedReadsAreBoundedLivePostgresTests). + private static async Task PrepareHypertablesAsync(string connectionString, NpgsqlConnection connection, CancellationToken ct) + { + var timescale = await LiveTimescaleProbe.TryEnableAsync(connectionString, ct); + if (timescale) + { + await TimescaleSupport.ConvertToHypertablesAsync(connection, null, ct); + await TimescaleSupport.EnsureCollectionLogHypertableAsync(connection, null, ct); + } + + return timescale; + } + + /// PostgreSQL timestamp is microsecond-resolution; .NET ticks are 100 ns. + private static DateTime Micro(DateTime value) => + DateTime.SpecifyKind(new DateTime(value.Ticks - (value.Ticks % 10)), DateTimeKind.Unspecified); + + private static async Task DeleteSentinelRowsAsync(NpgsqlConnection connection, CancellationToken ct) + { + var ids = string.Join(", ", AllNames.Select(n => Id(n).ToString(CultureInfo.InvariantCulture))); + foreach (var table in new[] { "ag_replica_states", "ag_database_replica_states" }) + { + using var cleanup = new NpgsqlCommand($"DELETE FROM {table} WHERE server_id IN ({ids})", connection); + await cleanup.ExecuteNonQueryAsync(ct); + } + + using (var cleanupConfig = new NpgsqlCommand($"DELETE FROM config_monitored_servers WHERE server_id IN ({ids})", connection)) + { + await cleanupConfig.ExecuteNonQueryAsync(ct); + } + + using var cleanupServers = new NpgsqlCommand($"DELETE FROM servers WHERE server_id IN ({ids})", connection); + await cleanupServers.ExecuteNonQueryAsync(ct); + } +} diff --git a/Darling/Darling.Tests/DarlingAgStatesReaderTests.cs b/Darling/Darling.Tests/DarlingAgStatesReaderTests.cs new file mode 100644 index 0000000000..811648bcad --- /dev/null +++ b/Darling/Darling.Tests/DarlingAgStatesReaderTests.cs @@ -0,0 +1,298 @@ +/* + * 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.Generic; +using System.Linq; +using System.Reflection; +using System.Text.RegularExpressions; +using PerformanceMonitor.Darling.Storage; +using Xunit; + +namespace Darling.Tests; + +/// +/// The shared Availability Group reader (#4228), ungated: the Postgres dialect and shape of every statement it +/// ships, the source pin that none of them regresses to the unbounded shape this issue replaced, and the pure +/// staleness split that decides the shared floor. This is now the ONE place the Viewer's AG tab, /api/ag +/// and get_ag_health get their SQL from — and +/// no longer carry their own copies of these pins. +/// +public sealed class DarlingAgStatesReaderTests +{ + private static readonly string[] StepTwoStatements = + { + DarlingAgStatesReader.ReplicaStatesSql, + DarlingAgStatesReader.DatabaseReplicaStatesSql, + DarlingAgStatesReader.ReplicaGroupCountSql, + }; + + private static readonly string[] StepOneStatements = + { + DarlingAgStatesReader.ReplicaNewestInstantSql, + DarlingAgStatesReader.DatabaseNewestInstantSql, + }; + + /* ─────────────────────────── source pin ─────────────────────────── */ + + /// The issue's own pin (#4228): no statement this reader ships may run + /// MAX(collection_time) ... GROUP BY server_id with nothing bounding collection_time between + /// the FROM and the GROUP BY — that shape is what read every retained row for every server (2,803 ms, + /// 441 k buffers for 398 rows). Reflects over every public string constant on the type, so a future + /// statement pasted in the old shape fails here instead of shipping the regression again. + [Fact] + public void NoStatement_RunsTheUnboundedNewestSnapshotAggregate() + { + var unbounded = new Regex( + @"FROM\s+ag_(replica_states|database_replica_states)\s*\r?\n\s*GROUP BY server_id", + RegexOptions.IgnoreCase); + + var constants = typeof(DarlingAgStatesReader) + .GetFields(BindingFlags.Public | BindingFlags.Static) + .Where(f => f.IsLiteral && f.FieldType == typeof(string)); + + var checkedAny = false; + foreach (var field in constants) + { + var sql = (string)field.GetValue(null)!; + checkedAny = true; + Assert.False( + unbounded.IsMatch(sql), + $"{field.Name} runs MAX(collection_time)/GROUP BY server_id with no collection_time floor (#4228)."); + } + + Assert.True(checkedAny, "no public SQL constants found to check — the reflection scope is wrong."); + } + + /* ─────────────────────────── step one: dialect + shape ─────────────────────────── */ + + [Fact] + public void StepOneSql_IsPostgresDialect_WithPositionalServerFilter() + { + foreach (var sql in StepOneStatements) + { + Assert.Contains("$1", sql, StringComparison.Ordinal); + Assert.DoesNotContain("@", sql, StringComparison.Ordinal); + Assert.DoesNotContain("N'", sql, StringComparison.Ordinal); + Assert.Equal(1, Occurrences(sql, "$1::integer IS NULL")); + } + } + + [Fact] + public void StepOneSql_IsAnOrderedDescentPerServer_NotAnAggregate() + { + foreach (var sql in StepOneStatements) + { + Assert.Contains("ORDER BY", sql, StringComparison.Ordinal); + Assert.Contains("DESC", sql, StringComparison.Ordinal); + Assert.Contains("LIMIT 1", sql, StringComparison.Ordinal); + Assert.DoesNotContain("MAX(collection_time)", sql, StringComparison.Ordinal); + Assert.DoesNotContain("GROUP BY", sql, StringComparison.Ordinal); + } + } + + /// The ruling this issue adds on top of its own sketch: step one is driven from EVERY row of + /// servers, not scoped to is_enabled. #4236 was a by-server read scoped to an enabled table + /// that dropped a disabled server's own rows when looked up directly; step one plays that same + /// per-server-lookup role for the floor/staleness split, so it stays unscoped and lets the OUTPUT-side + /// join (unchanged, still present below) decide visibility, exactly as today. + [Fact] + public void StepOneSql_IsNotScopedToEnabledServers() + { + foreach (var sql in StepOneStatements) + { + Assert.DoesNotContain("is_enabled", sql, StringComparison.Ordinal); + } + } + + [Fact] + public void StepOneSql_ReadsTheBareTables() + { + Assert.Contains("FROM ag_replica_states", DarlingAgStatesReader.ReplicaNewestInstantSql, StringComparison.Ordinal); + Assert.Contains("FROM ag_database_replica_states", DarlingAgStatesReader.DatabaseNewestInstantSql, StringComparison.Ordinal); + + foreach (var sql in StepOneStatements.Concat(StepTwoStatements)) + { + Assert.DoesNotContain("collect.", sql, StringComparison.Ordinal); + } + } + + /* ─────────────────────────── step two: dialect + shape ─────────────────────────── */ + + [Fact] + public void StepTwoSql_IsPostgresDialect_WithPositionalFloorAndServerFilter() + { + foreach (var sql in StepTwoStatements) + { + Assert.DoesNotContain("@", sql, StringComparison.Ordinal); + Assert.DoesNotContain("N'", sql, StringComparison.Ordinal); + + /* The floor guards BOTH the inner latest-instant aggregate and the outer scan — count, don't just + Contains, or deleting either WHERE leaves this green (#4228's whole point). */ + Assert.Equal(2, Occurrences(sql, "collection_time >= $1")); + + /* The optional server filter guards both arms too, exactly as the shipped Service statement did + before the move (renumbered to $2 now that $1 is the floor). */ + Assert.Equal(2, Occurrences(sql, "$2::integer IS NULL")); + } + } + + [Fact] + public void StepTwoSql_KeepsEveryRowAtEachServersNewestCollection() + { + /* DISTINCT ON would keep ONE row per server — wrong here, since a snapshot is many rows (one per + replica / per database). The join against MAX(collection_time) keeps them all; unchanged by the + floor bound added alongside it. */ + foreach (var sql in new[] { DarlingAgStatesReader.ReplicaStatesSql, DarlingAgStatesReader.DatabaseReplicaStatesSql }) + { + Assert.Contains("MAX(collection_time)", sql, StringComparison.Ordinal); + Assert.Contains("GROUP BY server_id", sql, StringComparison.Ordinal); + Assert.DoesNotContain("DISTINCT ON", sql, StringComparison.Ordinal); + } + } + + [Fact] + public void StepTwoSql_RestrictsToTheEnabledServerRegistry() + { + /* Unchanged from the shipped statements: a server disabled in the control plane leaves the fleet + surfaces at once instead of showing stale AG cards until its rows age out. */ + foreach (var sql in StepTwoStatements) + { + Assert.Contains("JOIN servers AS s", sql, StringComparison.Ordinal); + Assert.Contains("s.is_enabled", sql, StringComparison.Ordinal); + } + } + + [Fact] + public void StepTwoSql_SelectsIsLocalOnBothGrains() + { + Assert.Contains("r.is_local", DarlingAgStatesReader.ReplicaStatesSql, StringComparison.Ordinal); + Assert.Contains("d.is_local", DarlingAgStatesReader.DatabaseReplicaStatesSql, StringComparison.Ordinal); + } + + [Fact] + public void StepTwoSql_SelectsEveryColumnTheDtosSurface() + { + foreach (var column in new[] + { + "role_desc", "operational_state_desc", "connected_state_desc", "recovery_health_desc", + "synchronization_health_desc", "availability_mode_desc", "failover_mode_desc", "endpoint_url", + }) + { + Assert.Contains(column, DarlingAgStatesReader.ReplicaStatesSql, StringComparison.Ordinal); + } + + foreach (var column in new[] + { + "synchronization_state_desc", "last_hardened_lsn", "last_commit_lsn", "log_send_queue_size", + "redo_queue_size", "log_send_rate", "redo_rate", "is_suspended", "suspend_reason_desc", + "secondary_lag_seconds", "is_local", + }) + { + Assert.Contains(column, DarlingAgStatesReader.DatabaseReplicaStatesSql, StringComparison.Ordinal); + } + } + + private static int Occurrences(string haystack, string needle) + { + var count = 0; + for (var i = haystack.IndexOf(needle, StringComparison.Ordinal); i >= 0; + i = haystack.IndexOf(needle, i + needle.Length, StringComparison.Ordinal)) + { + count++; + } + + return count; + } + + /* ─────────────────────────── the staleness split (pure) ─────────────────────────── */ + + [Fact] + public void SplitCurrentAndStale_EmptyInput_YieldsNoFloorAndNoStale() + { + var (floor, stale) = DarlingAgStatesReader.SplitCurrentAndStale( + Array.Empty<(int ServerId, DateTime Instant)>(), DateTime.UtcNow); + + Assert.Null(floor); + Assert.Empty(stale); + } + + [Fact] + public void SplitCurrentAndStale_EveryServerCurrent_FloorIsTheMinimum_NoneStale() + { + var now = new DateTime(2026, 9, 25, 12, 0, 0, DateTimeKind.Utc); + var instants = new List<(int ServerId, DateTime Instant)> + { + (1, now.AddMinutes(-5)), + (2, now.AddMinutes(-30)), + (3, now), + }; + + var (floor, stale) = DarlingAgStatesReader.SplitCurrentAndStale(instants, now); + + Assert.Equal(now.AddMinutes(-30), floor); + Assert.Empty(stale); + } + + /// The scenario #4228 exists to fix: one server's AG rows stopped a day ago. Its instant must not + /// become the shared floor — it is read on its own instead, and the shared floor tracks only the servers + /// still collecting. + [Fact] + public void SplitCurrentAndStale_OneServerStoppedADayAgo_IsCarvedOutAndDoesNotDragTheFloor() + { + var now = new DateTime(2026, 9, 25, 12, 0, 0, DateTimeKind.Utc); + var instants = new List<(int ServerId, DateTime Instant)> + { + (1, now.AddMinutes(-5)), + (2, now.AddDays(-1).AddMinutes(-1)), + }; + + var (floor, stale) = DarlingAgStatesReader.SplitCurrentAndStale(instants, now); + + Assert.Equal(now.AddMinutes(-5), floor); + var staleServer = Assert.Single(stale); + Assert.Equal(2, staleServer.ServerId); + Assert.Equal(now.AddDays(-1).AddMinutes(-1), staleServer.Instant); + } + + [Fact] + public void SplitCurrentAndStale_EveryServerStale_NoFloor_EveryServerReadOnItsOwn() + { + var now = new DateTime(2026, 9, 25, 12, 0, 0, DateTimeKind.Utc); + var instants = new List<(int ServerId, DateTime Instant)> + { + (1, now.AddDays(-3)), + (2, now.AddDays(-10)), + }; + + var (floor, stale) = DarlingAgStatesReader.SplitCurrentAndStale(instants, now); + + Assert.Null(floor); + Assert.Equal(2, stale.Count); + } + + /// The horizon boundary: exactly at is stale + /// (the split uses a strict less-than on the horizon START), one tick inside it is current. Pins the + /// edge rather than leaving it to whichever side floating-point-free DateTime math happens to fall. + [Fact] + public void SplitCurrentAndStale_HorizonBoundary_IsExclusive() + { + var now = new DateTime(2026, 9, 25, 12, 0, 0, DateTimeKind.Utc); + var horizonStart = now - DarlingAgStatesReader.StalenessHorizon; + + var atBoundary = DarlingAgStatesReader.SplitCurrentAndStale( + new List<(int ServerId, DateTime Instant)> { (1, horizonStart) }, now); + Assert.Equal(horizonStart, atBoundary.Floor); + Assert.Empty(atBoundary.Stale); + + var justStale = DarlingAgStatesReader.SplitCurrentAndStale( + new List<(int ServerId, DateTime Instant)> { (1, horizonStart.AddTicks(-1)) }, now); + Assert.Null(justStale.Floor); + Assert.Single(justStale.Stale); + } +} diff --git a/Darling/Darling.Tests/ViewerAvailabilityGroupsTests.cs b/Darling/Darling.Tests/ViewerAvailabilityGroupsTests.cs index 08b1def51b..ad50eb7062 100644 --- a/Darling/Darling.Tests/ViewerAvailabilityGroupsTests.cs +++ b/Darling/Darling.Tests/ViewerAvailabilityGroupsTests.cs @@ -15,46 +15,19 @@ namespace Darling.Tests; /// -/// The viewer's Availability Groups tab (#991): the read SQL's dialect and latest-snapshot shape, the banding -/// rules, and the pure card projection. Ungated — the projection half is separated from the Postgres half exactly -/// so the rules test without a store, mirroring FleetViewTests. -/// -/// These deliberately duplicate DarlingAgReaderTests' expectations rather than sharing them: the -/// viewer reader is a COPY of the service reader (no ProjectReference exists), so the pins are what keep the two -/// from drifting into disagreeing about whether an AG is healthy. +/// The viewer's Availability Groups tab (#991): the banding rules and the pure card projection. Ungated — this +/// half is separated from the Postgres half exactly so the rules test without a store, mirroring FleetViewTests. +/// The read SQL's dialect and latest-snapshot shape are pinned once, in DarlingAgStatesReaderTests +/// (#4228) — this file and DarlingAgReaderTests both call the SAME DarlingAgStatesReader now, so +/// there is one statement to disagree about instead of two. /// public sealed class AgTopologyCardsTests { - /* ─────────────────────────── SQL pins ─────────────────────────── */ - - [Fact] - public void AgSql_IsPostgresDialect_AndReadsTheBareCollectTables() - { - foreach (var sql in new[] { ViewerDataService.AgReplicaStatesSql, ViewerDataService.AgDatabaseReplicaStatesSql }) - { - /* The viewer's store connection resolves `collect` through search_path, so these must NOT be - schema-qualified the way the service reader's copies are. */ - Assert.DoesNotContain("collect.", sql, StringComparison.Ordinal); - Assert.DoesNotContain("@", sql, StringComparison.Ordinal); - Assert.DoesNotContain("N'", sql, StringComparison.Ordinal); - } - - Assert.Contains("FROM ag_replica_states", ViewerDataService.AgReplicaStatesSql, StringComparison.Ordinal); - Assert.Contains("FROM ag_database_replica_states", ViewerDataService.AgDatabaseReplicaStatesSql, StringComparison.Ordinal); - } - - [Fact] - public void AgSql_KeepsEveryRowAtEachServersNewestCollection() - { - /* DISTINCT ON would keep ONE row per server, which is exactly wrong: a snapshot is many rows (one per - replica / per database). The join against MAX(collection_time) keeps them all. */ - foreach (var sql in new[] { ViewerDataService.AgReplicaStatesSql, ViewerDataService.AgDatabaseReplicaStatesSql }) - { - Assert.Contains("MAX(collection_time)", sql, StringComparison.Ordinal); - Assert.Contains("GROUP BY server_id", sql, StringComparison.Ordinal); - Assert.DoesNotContain("DISTINCT ON", sql, StringComparison.Ordinal); - } - } + /* The SQL pins used to live here, against this file's own copy of the statement text (the viewer had no + route to the service assembly that carried the other copy). Both copies moved to DarlingAgStatesReader + in PerformanceMonitor.Darling.Storage (#4228), which this project already references, and the dialect / + shape pins moved with it to DarlingAgStatesReaderTests — one set of pins for the one implementation, + instead of two that could drift apart. */ [Theory] [InlineData(true, "local")] @@ -77,28 +50,6 @@ public void LocalDisplay_ShownOnlyWhenExplicitlyLocal(bool? isLocal, string expe Assert.Equal(expected, Assert.Single(card.Replicas).LocalDisplay); } - [Fact] - public void AgSql_SelectsIsLocalOnBothGrains() - { - /* is_local arrived on the replica grain in V37; the database grain has carried it since V34. Both reads - must select it, and the replica read's ordinals shifted when it was inserted — this pin plus the - projection tests are what stand between that and a silently mis-mapped column. */ - Assert.Contains("r.is_local", ViewerDataService.AgReplicaStatesSql, StringComparison.Ordinal); - Assert.Contains("d.is_local", ViewerDataService.AgDatabaseReplicaStatesSql, StringComparison.Ordinal); - } - - [Fact] - public void AgSql_RestrictsToTheEnabledServerRegistry() - { - /* A server disabled in the control plane leaves the fleet surfaces at once; its AG cards go with it - rather than lingering until retention expires the rows. */ - foreach (var sql in new[] { ViewerDataService.AgReplicaStatesSql, ViewerDataService.AgDatabaseReplicaStatesSql }) - { - Assert.Contains("JOIN servers AS s", sql, StringComparison.Ordinal); - Assert.Contains("s.is_enabled", sql, StringComparison.Ordinal); - } - } - /* ─────────────────────────── banding ─────────────────────────── */ [Theory] diff --git a/Darling/PerformanceMonitor.Darling.Service/Mcp/DarlingAgReader.cs b/Darling/PerformanceMonitor.Darling.Service/Mcp/DarlingAgReader.cs index 00a3ae06a9..e825873a6a 100644 --- a/Darling/PerformanceMonitor.Darling.Service/Mcp/DarlingAgReader.cs +++ b/Darling/PerformanceMonitor.Darling.Service/Mcp/DarlingAgReader.cs @@ -8,7 +8,6 @@ using System; using System.Collections.Generic; -using System.Globalization; using System.Linq; using System.Text.Json; using System.Text.Json.Serialization; @@ -16,6 +15,7 @@ using System.Threading.Tasks; using Npgsql; using PerformanceMonitor.Common; +using PerformanceMonitor.Darling.Storage; namespace PerformanceMonitor.Darling.Service.Mcp; @@ -34,10 +34,11 @@ namespace PerformanceMonitor.Darling.Service.Mcp; /// /// Latest collection per server. Each server contributes only the rows from its newest AG collection /// (the JOIN ... MAX(collection_time) GROUP BY server_id shape DarlingFleetReader uses for its -/// latest-snapshot reads). Because the collectors write NO row when a server has no AGs, a server whose AGs were -/// dropped keeps returning its last non-empty snapshot — so every group carries its -/// and the UI shows it, rather than the reader guessing a -/// staleness threshold. +/// latest-snapshot reads, now floor-bounded — see , #4228). Because the +/// collectors write NO row when a server has no AGs, a server whose AGs were dropped keeps returning its last +/// non-empty snapshot no matter how old — so every group carries its +/// and the UI shows it. only picks HOW that server's row gets +/// fetched (its own point read instead of feeding the shared floor); it never gates WHETHER the row comes back. /// /// Banding lives here (R1). Every severity below is derived server-side and serialized with the /// result; the browser only maps a severity NAME to a CSS class and never re-derives a threshold. The mappings are @@ -66,108 +67,14 @@ internal static class DarlingAgReader Converters = { new JsonStringEnumConverter() }, }; - /* ─────────────────────────── SQL (public for dialect pinning) ─────────────────────────── */ - - /// Every replica row from each server's NEWEST AG collection. The inner aggregate picks the latest - /// instant per server and the join keeps ALL rows at that instant (not DISTINCT ON, which would keep - /// only one replica). Joined to the ENABLED registry, like every other fleet read — a server disabled in the - /// control plane leaves the fleet surfaces at once instead of showing stale AG cards until its rows age out. - /// $1 an optional server_id filter — NULL means the whole fleet. - public const string ReplicaStatesSql = @" -SELECT - r.server_id, - r.server_name, - r.collection_time, - r.ag_name, - r.replica_server_name, - r.role_desc, - r.is_local, - r.operational_state_desc, - r.connected_state_desc, - r.recovery_health_desc, - r.synchronization_health_desc, - r.availability_mode_desc, - r.failover_mode_desc, - r.endpoint_url -FROM collect.ag_replica_states AS r -JOIN -( - SELECT server_id, MAX(collection_time) AS max_collection_time - FROM collect.ag_replica_states - WHERE ($1::integer IS NULL OR server_id = $1) - GROUP BY server_id -) AS latest - ON r.server_id = latest.server_id - AND r.collection_time = latest.max_collection_time -JOIN servers AS s - ON s.server_id = r.server_id - AND s.is_enabled -WHERE ($1::integer IS NULL OR r.server_id = $1) -ORDER BY r.server_name, r.ag_name, r.replica_server_name"; - - /// Every database-grain row from each server's NEWEST AG collection, same latest-snapshot shape as - /// . Windowed independently of the replica read: the two collectors run as - /// separate sweeps, so their newest instants need not coincide. $1 an optional server_id filter — NULL means - /// the whole fleet. - public const string DatabaseReplicaStatesSql = @" -SELECT - d.server_id, - d.server_name, - d.collection_time, - d.ag_name, - d.database_name, - d.replica_server_name, - d.is_local, - d.synchronization_state_desc, - d.last_hardened_lsn, - d.last_commit_lsn, - d.log_send_queue_size, - d.redo_queue_size, - d.log_send_rate, - d.redo_rate, - d.is_suspended, - d.suspend_reason_desc, - d.availability_mode_desc, - d.secondary_lag_seconds -FROM collect.ag_database_replica_states AS d -JOIN -( - SELECT server_id, MAX(collection_time) AS max_collection_time - FROM collect.ag_database_replica_states - WHERE ($1::integer IS NULL OR server_id = $1) - GROUP BY server_id -) AS latest - ON d.server_id = latest.server_id - AND d.collection_time = latest.max_collection_time -JOIN servers AS s - ON s.server_id = d.server_id - AND s.is_enabled -WHERE ($1::integer IS NULL OR d.server_id = $1) -ORDER BY d.server_name, d.ag_name, d.database_name, d.replica_server_name"; - - /// The nav-gate probe's query (#4189): the SAME (server, ag) groups - /// would build, counted server-side instead of assembled into rows. Only the replica grain — 's - /// group set is defined entirely by the replica rows (a database row whose group has no replica row is - /// dropped), so the database-grain table never changes this count and this query does not join it. - /// UPPER(COALESCE(r.ag_name, '')) mirrors 's case fold and its NULL-to-empty-string - /// mapping exactly, so this can never disagree with AvailabilityGroupCount for the same snapshot — - /// pinned by DarlingAgReaderTests. $1 an optional server_id filter — NULL means the whole fleet. - public const string AvailabilityGroupCountSql = @" -SELECT COUNT(DISTINCT (r.server_id, UPPER(COALESCE(r.ag_name, '')))) -FROM collect.ag_replica_states AS r -JOIN -( - SELECT server_id, MAX(collection_time) AS max_collection_time - FROM collect.ag_replica_states - WHERE ($1::integer IS NULL OR server_id = $1) - GROUP BY server_id -) AS latest - ON r.server_id = latest.server_id - AND r.collection_time = latest.max_collection_time -JOIN servers AS s - ON s.server_id = r.server_id - AND s.is_enabled -WHERE ($1::integer IS NULL OR r.server_id = $1)"; + /* ─────────────────────────── SQL ─────────────────────────── */ + + /// The statement text and the two-step, floor-bounded read live in + /// (#4228) — the Viewer's AG tab, /api/ag and this tool all call + /// the SAME implementation now, so there is one place, not three, that can disagree about what "the newest + /// snapshot" means. / below map its raw rows + /// into this file's / , unchanged from before the move, so + /// and everything downstream of it needed no edit. /* ─────────────────────────── the read ─────────────────────────── */ @@ -182,36 +89,31 @@ public static async Task GetAgHealthAsync( DateTime? nowUtc = null, CancellationToken cancellationToken = default) { - var replicas = await ReadReplicasAsync(postgres, serverIdFilter, cancellationToken); + var effectiveNow = nowUtc ?? DateTime.UtcNow; + var replicas = await ReadReplicasAsync(postgres, serverIdFilter, effectiveNow, cancellationToken); /* Short-circuit: no replica rows means no AGs anywhere in scope, and the database-grain table cannot hold rows for an AG that has no replicas. The common case (a fleet with no Always On at all) therefore costs exactly one indexed read that returns nothing — this read runs on every dashboard poll. */ var databases = replicas.Count == 0 ? new List() - : await ReadDatabasesAsync(postgres, serverIdFilter, cancellationToken); + : await ReadDatabasesAsync(postgres, serverIdFilter, effectiveNow, cancellationToken); - return Build(replicas, databases, nowUtc ?? DateTime.UtcNow); + return Build(replicas, databases, effectiveNow); } /// /// The nav-gate probe (#4189): "how many availability groups does this scope have", without building a /// single to answer it — no per-replica columns, no ORDER BY, no - /// database-grain read. One aggregate query () in place of the + /// database-grain read. in place of the /// topology read runs; that read's own AvailabilityGroupCount is /// exactly this number for the same scope and instant. /// - public static async Task GetAvailabilityGroupCountAsync( + public static Task GetAvailabilityGroupCountAsync( NpgsqlDataSource postgres, int? serverIdFilter = null, - CancellationToken cancellationToken = default) - { - await using var command = postgres.CreateCommand(AvailabilityGroupCountSql); - command.CommandTimeout = McpCommandDeadlines.ReadSeconds; - AddServerFilter(command, serverIdFilter); - var result = await command.ExecuteScalarAsync(cancellationToken); - return Convert.ToInt32(result ?? 0L); - } + CancellationToken cancellationToken = default) => + DarlingAgStatesReader.GetReplicaGroupCountAsync(postgres, serverIdFilter, DateTime.UtcNow, cancellationToken); /// /// Assembles the raw rows into per-(server, AG) groups — pure, so the grouping and every band unit-test @@ -507,88 +409,43 @@ private static string Normalize(string? value) => /* ─────────────────────────── reads ─────────────────────────── */ + /// Maps 's raw rows (#4228) into this file's own + /// , field for field, so and every banding rule below it needed + /// no change when the read moved. private static async Task> ReadReplicasAsync( - NpgsqlDataSource postgres, int? serverIdFilter, CancellationToken cancellationToken) + NpgsqlDataSource postgres, int? serverIdFilter, DateTime nowUtc, CancellationToken cancellationToken) { - var rows = new List(); - await using var command = postgres.CreateCommand(ReplicaStatesSql); - command.CommandTimeout = McpCommandDeadlines.ReadSeconds; - AddServerFilter(command, serverIdFilter); - await using var reader = await command.ExecuteReaderAsync(cancellationToken); - while (await reader.ReadAsync(cancellationToken)) + var raw = await DarlingAgStatesReader.GetReplicaStatesAsync(postgres, serverIdFilter, nowUtc, cancellationToken); + var rows = new List(raw.Count); + foreach (var row in raw) { rows.Add(new ReplicaRow( - reader.GetInt32(0), - reader.GetString(1), - reader.GetDateTime(2), - Text(reader, 3), - Text(reader, 4), - Text(reader, 5), - Flag(reader, 6), - Text(reader, 7), - Text(reader, 8), - Text(reader, 9), - Text(reader, 10), - Text(reader, 11), - Text(reader, 12), - Text(reader, 13))); + row.ServerId, row.ServerName, row.CollectionTime, row.AgName, row.ReplicaServerName, row.RoleDesc, + row.IsLocal, row.OperationalStateDesc, row.ConnectedStateDesc, row.RecoveryHealthDesc, + row.SynchronizationHealthDesc, row.AvailabilityModeDesc, row.FailoverModeDesc, row.EndpointUrl)); } return rows; } + /// Same mapping as , for the database grain. private static async Task> ReadDatabasesAsync( - NpgsqlDataSource postgres, int? serverIdFilter, CancellationToken cancellationToken) + NpgsqlDataSource postgres, int? serverIdFilter, DateTime nowUtc, CancellationToken cancellationToken) { - var rows = new List(); - await using var command = postgres.CreateCommand(DatabaseReplicaStatesSql); - command.CommandTimeout = McpCommandDeadlines.ReadSeconds; - AddServerFilter(command, serverIdFilter); - await using var reader = await command.ExecuteReaderAsync(cancellationToken); - while (await reader.ReadAsync(cancellationToken)) + var raw = await DarlingAgStatesReader.GetDatabaseReplicaStatesAsync(postgres, serverIdFilter, nowUtc, cancellationToken); + var rows = new List(raw.Count); + foreach (var row in raw) { rows.Add(new DatabaseRow( - reader.GetInt32(0), - reader.GetString(1), - reader.GetDateTime(2), - Text(reader, 3), - Text(reader, 4), - Text(reader, 5), - Flag(reader, 6), - Text(reader, 7), - Text(reader, 8), - Text(reader, 9), - Count(reader, 10), - Count(reader, 11), - Count(reader, 12), - Count(reader, 13), - Flag(reader, 14), - Text(reader, 15), - Text(reader, 16), - Count(reader, 17))); + row.ServerId, row.ServerName, row.CollectionTime, row.AgName, row.DatabaseName, row.ReplicaServerName, + row.IsLocal, row.SynchronizationStateDesc, row.LastHardenedLsn, row.LastCommitLsn, row.LogSendQueueSize, + row.RedoQueueSize, row.LogSendRate, row.RedoRate, row.IsSuspended, row.SuspendReasonDesc, + row.AvailabilityModeDesc, row.SecondaryLagSeconds)); } return rows; } - /// Binds the optional server filter as a nullable integer — NULL means the whole fleet, which is what - /// the SQL's $1::integer IS NULL OR ... guard reads. - private static void AddServerFilter(NpgsqlCommand command, int? serverIdFilter) => - command.Parameters.Add(new NpgsqlParameter - { - NpgsqlDbType = NpgsqlTypes.NpgsqlDbType.Integer, - Value = (object?)serverIdFilter ?? DBNull.Value, - }); - - private static string? Text(NpgsqlDataReader reader, int ordinal) => - reader.IsDBNull(ordinal) ? null : reader.GetString(ordinal); - - private static bool? Flag(NpgsqlDataReader reader, int ordinal) => - reader.IsDBNull(ordinal) ? null : reader.GetBoolean(ordinal); - - private static long? Count(NpgsqlDataReader reader, int ordinal) => - reader.IsDBNull(ordinal) ? null : Convert.ToInt64(reader.GetValue(ordinal), CultureInfo.InvariantCulture); - /* ─────────────────────────── raw-read carriers ─────────────────────────── */ /// One replica-grain row, exactly as collect.ag_replica_states stores it. diff --git a/Darling/PerformanceMonitor.Darling.Storage/DarlingAgStatesReader.cs b/Darling/PerformanceMonitor.Darling.Storage/DarlingAgStatesReader.cs new file mode 100644 index 0000000000..6ea4db5ef6 --- /dev/null +++ b/Darling/PerformanceMonitor.Darling.Storage/DarlingAgStatesReader.cs @@ -0,0 +1,533 @@ +/* + * 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.Generic; +using System.Threading; +using System.Threading.Tasks; +using Npgsql; +using NpgsqlTypes; + +namespace PerformanceMonitor.Darling.Storage; + +/// +/// The Availability Group topology reads (#991) behind the WPF viewer's AG tab, the web dashboard's +/// /api/ag and /api/ag/count, and the get_ag_health MCP tool — moved here (#4228), like +/// before it, because the Viewer has no route to the Service +/// assembly and three surfaces were carrying independent copies of the same statement text. +/// +/// The cost problem this replaces (#4228). Every shipped statement found "each server's newest +/// AG snapshot" by joining to SELECT server_id, MAX(collection_time) ... GROUP BY server_id with no +/// bound on collection_time. That inner aggregate reads every retained row for every server — +/// 2,803 ms and 441 k buffers for 398 rows on a production store — because the instant it finds is not known +/// until the whole table has been read, so TimescaleDB cannot exclude a single chunk. A literal lower bound +/// on the partition column is what makes the read cheap (2.1 ms + 1.0 ms measured for the same 398 rows), so +/// every read below is TWO statements: / +/// find each server's newest instant by an ORDERED DESCENT per server +/// (index range scan, one chunk), and / +/// / re-run the original shape with that instant as a literal floor on +/// BOTH the inner aggregate and the outer scan. +/// +/// Why the floor is a shared MINIMUM, not one server_id at a time. The two-step measurement +/// that hit 1.0 ms ran ONE statement for the whole fleet with one floor, not a per-server LATERAL join back +/// (that form was measured too: 2,090 ms, because probing every chunk for a found instant is still probing +/// every chunk). So step two stays the fleet-wide statement it always was; only the floor changes. +/// +/// Stale servers do not drag the shared floor back (#4228). A server whose AG collection +/// stopped days ago would otherwise pull the shared floor back to its own ancient instant, reintroducing the +/// full scan for every OTHER server too. — one chunk width — splits step +/// one's per-server instants into CURRENT (within the horizon, feed the shared floor, read together in one +/// statement) and STALE (older, read on its OWN with its OWN instant as the floor — a single-chunk point +/// read, same statement text, just scoped to one server_id). Results are identical to the unbounded shape +/// either way; only the read plan differs. One chunk width keeps the shared floor inside "today plus +/// yesterday" whenever every server is current, which is what bounds the outer scan to at most two chunks — +/// a wider horizon would let the shared floor span three. +/// +/// Step one is NOT scoped to servers.is_enabled, on purpose, even though the issue's own +/// sketch shows that filter. #4236 (the same night) was a by-server read scoped to an enabled/registered +/// table, which dropped a disabled server's and an unregistered server's own rows when a caller looked THAT +/// server up directly. Step one plays the same role here — it decides what feeds the floor and what gets its +/// own point read, not what is VISIBLE. Every statement below keeps the SAME JOIN servers AS s ON +/// s.server_id = ... AND s.is_enabled the shipped statements always had, unchanged, so a disabled +/// server's rows are excluded from the OUTPUT exactly as they are today, and a server missing from +/// servers entirely (no FK ties ag_replica_states.server_id to it) is excluded the same way +/// today's INNER JOIN already excludes it. Scoping step one to enabled servers would not change either of +/// those outcomes — the output-side join already decides them — so the only effect of adding it would be to +/// risk repeating #4236's mistake for no benefit. +/// +/// Bare table names, matching . The store session's +/// search_path is collect, config, public (), so +/// ag_replica_states and ag_database_replica_states (both collect-schema) and +/// servers (public-schema) all resolve unqualified. The Viewer's copy of these statements was already +/// bare; the Service's copy was collect.-qualified. This picks the Storage-layer convention. +/// +/// The database grain selects two columns the Viewer does not currently show +/// (last_hardened_lsn, last_commit_lsn) because the Service's shipped statement already +/// selected them and one shared statement has one column list. The Viewer's mapping simply does not read +/// them, exactly as it does not today. +/// +public static class DarlingAgStatesReader +{ + /// How far a server's newest AG snapshot may fall behind nowUtc and still feed the + /// shared floor. One chunk width (): with every server + /// current, the shared floor is at most one chunk width old, so the bounded outer scan touches at most + /// two calendar-day chunks (today's, plus yesterday's when the floor falls before UTC midnight) — the + /// live EXPLAIN pin's bound (#4228). + public static readonly TimeSpan StalenessHorizon = TimeSpan.FromDays(TimescaleSupport.ChunkIntervalDays); + + /* ─────────────────────────── SQL: step one, per-server newest instant ─────────────────────────── */ + + /// Step one for the replica grain: every server's newest ag_replica_states instant, via + /// an ordered descent per server (the index on (server_id, collection_time) answers each LATERAL + /// with a single-chunk range scan) rather than the unbounded MAX(...) GROUP BY this replaces. + /// Driven from servers so the driving set is fleet-sized, not retention-sized; deliberately NOT + /// scoped to is_enabled (see the type doc). A server with no replica rows contributes no row (CROSS + /// JOIN LATERAL). $1 an optional server_id filter — NULL means the whole fleet. + public const string ReplicaNewestInstantSql = """ + SELECT s.server_id, l.collection_time + FROM servers AS s + CROSS JOIN LATERAL + ( + SELECT r.collection_time + FROM ag_replica_states AS r + WHERE r.server_id = s.server_id + ORDER BY r.collection_time DESC + LIMIT 1 + ) AS l + WHERE ($1::integer IS NULL OR s.server_id = $1) + """; + + /// Step one for the database grain — same shape as , over + /// ag_database_replica_states. Windowed independently: the two collectors sweep separately, so + /// their newest instants need not coincide, exactly as the shipped statements always assumed. $1 an + /// optional server_id filter — NULL means the whole fleet. + public const string DatabaseNewestInstantSql = """ + SELECT s.server_id, l.collection_time + FROM servers AS s + CROSS JOIN LATERAL + ( + SELECT d.collection_time + FROM ag_database_replica_states AS d + WHERE d.server_id = s.server_id + ORDER BY d.collection_time DESC + LIMIT 1 + ) AS l + WHERE ($1::integer IS NULL OR s.server_id = $1) + """; + + /* ─────────────────────────── SQL: step two, floor-bounded snapshot ─────────────────────────── */ + + /// Step two for the replica grain: the shipped statement's exact shape (the inner aggregate, + /// the join back on the found instant, the enabled-registry join — all unchanged), with + /// collection_time >= $1 added to BOTH the inner aggregate and the outer scan. Called once for + /// the whole "current" set with the shared floor, and once per "stale" server with that server's own + /// instant as $1 and its own server_id as $2. $1 the floor (naive UTC); $2 an optional server_id filter — + /// NULL means every current server. + public const string ReplicaStatesSql = """ + SELECT + r.server_id, + r.server_name, + r.collection_time, + r.ag_name, + r.replica_server_name, + r.role_desc, + r.is_local, + r.operational_state_desc, + r.connected_state_desc, + r.recovery_health_desc, + r.synchronization_health_desc, + r.availability_mode_desc, + r.failover_mode_desc, + r.endpoint_url + FROM ag_replica_states AS r + JOIN + ( + SELECT server_id, MAX(collection_time) AS max_collection_time + FROM ag_replica_states + WHERE collection_time >= $1 + AND ($2::integer IS NULL OR server_id = $2) + GROUP BY server_id + ) AS latest + ON r.server_id = latest.server_id + AND r.collection_time = latest.max_collection_time + JOIN servers AS s + ON s.server_id = r.server_id + AND s.is_enabled + WHERE r.collection_time >= $1 + AND ($2::integer IS NULL OR r.server_id = $2) + ORDER BY r.server_name, r.ag_name, r.replica_server_name + """; + + /// Step two for the database grain — same shape as , over + /// ag_database_replica_states, selecting the Service's superset column list (includes + /// last_hardened_lsn / last_commit_lsn; see the type doc). $1 the floor (naive UTC); $2 an + /// optional server_id filter — NULL means every current server. + public const string DatabaseReplicaStatesSql = """ + SELECT + d.server_id, + d.server_name, + d.collection_time, + d.ag_name, + d.database_name, + d.replica_server_name, + d.is_local, + d.synchronization_state_desc, + d.last_hardened_lsn, + d.last_commit_lsn, + d.log_send_queue_size, + d.redo_queue_size, + d.log_send_rate, + d.redo_rate, + d.is_suspended, + d.suspend_reason_desc, + d.availability_mode_desc, + d.secondary_lag_seconds + FROM ag_database_replica_states AS d + JOIN + ( + SELECT server_id, MAX(collection_time) AS max_collection_time + FROM ag_database_replica_states + WHERE collection_time >= $1 + AND ($2::integer IS NULL OR server_id = $2) + GROUP BY server_id + ) AS latest + ON d.server_id = latest.server_id + AND d.collection_time = latest.max_collection_time + JOIN servers AS s + ON s.server_id = d.server_id + AND s.is_enabled + WHERE d.collection_time >= $1 + AND ($2::integer IS NULL OR d.server_id = $2) + ORDER BY d.server_name, d.ag_name, d.database_name, d.replica_server_name + """; + + /// The nav-gate probe's aggregate (#4189): the same (server, ag) groups + /// would build, counted server-side instead of assembled into rows. Same + /// floor-bounded shape as the row reads. $1 the floor (naive UTC); $2 an optional server_id filter — NULL + /// means every current server. + public const string ReplicaGroupCountSql = """ + SELECT COUNT(DISTINCT (r.server_id, UPPER(COALESCE(r.ag_name, '')))) + FROM ag_replica_states AS r + JOIN + ( + SELECT server_id, MAX(collection_time) AS max_collection_time + FROM ag_replica_states + WHERE collection_time >= $1 + AND ($2::integer IS NULL OR server_id = $2) + GROUP BY server_id + ) AS latest + ON r.server_id = latest.server_id + AND r.collection_time = latest.max_collection_time + JOIN servers AS s + ON s.server_id = r.server_id + AND s.is_enabled + WHERE r.collection_time >= $1 + AND ($2::integer IS NULL OR r.server_id = $2) + """; + + /* ─────────────────────────── rows ─────────────────────────── */ + + /// One replica-grain row, exactly as ag_replica_states stores it. + public readonly record struct ReplicaStateRow( + int ServerId, + string ServerName, + DateTime CollectionTime, + string? AgName, + string? ReplicaServerName, + string? RoleDesc, + bool? IsLocal, + string? OperationalStateDesc, + string? ConnectedStateDesc, + string? RecoveryHealthDesc, + string? SynchronizationHealthDesc, + string? AvailabilityModeDesc, + string? FailoverModeDesc, + string? EndpointUrl); + + /// One database-grain row, exactly as ag_database_replica_states stores it. Queue sizes + /// are KB and rates KB/s (the DMV's units), both instantaneous gauges rather than counters. + public readonly record struct DatabaseReplicaStateRow( + int ServerId, + string ServerName, + DateTime CollectionTime, + string? AgName, + string? DatabaseName, + string? ReplicaServerName, + bool? IsLocal, + string? SynchronizationStateDesc, + string? LastHardenedLsn, + string? LastCommitLsn, + long? LogSendQueueSize, + long? RedoQueueSize, + long? LogSendRate, + long? RedoRate, + bool? IsSuspended, + string? SuspendReasonDesc, + string? AvailabilityModeDesc, + long? SecondaryLagSeconds); + + /* ─────────────────────────── reads ─────────────────────────── */ + + /// Every replica row from each server's newest ag_replica_states snapshot — the two-step, + /// floor-bounded read (#4228). anchors the staleness split; pass the same + /// instant a caller would have stamped the unbounded read with. + public static async Task> GetReplicaStatesAsync( + NpgsqlDataSource postgres, int? serverIdFilter, DateTime nowUtc, CancellationToken cancellationToken = default) + { + var instants = await GetNewestInstantsAsync(postgres, ReplicaNewestInstantSql, serverIdFilter, cancellationToken); + var (floor, stale) = SplitCurrentAndStale(instants, nowUtc); + + var rows = new List(); + if (floor is not null) + { + rows.AddRange(await ReadReplicaStatesAsync(postgres, floor.Value, serverIdFilter, cancellationToken)); + } + + foreach (var server in stale) + { + rows.AddRange(await ReadReplicaStatesAsync(postgres, server.Instant, server.ServerId, cancellationToken)); + } + + rows.Sort(CompareReplicaRows); + return rows; + } + + /// Every database-grain row from each server's newest ag_database_replica_states snapshot + /// — same two-step shape as , windowed independently. + public static async Task> GetDatabaseReplicaStatesAsync( + NpgsqlDataSource postgres, int? serverIdFilter, DateTime nowUtc, CancellationToken cancellationToken = default) + { + var instants = await GetNewestInstantsAsync(postgres, DatabaseNewestInstantSql, serverIdFilter, cancellationToken); + var (floor, stale) = SplitCurrentAndStale(instants, nowUtc); + + var rows = new List(); + if (floor is not null) + { + rows.AddRange(await ReadDatabaseReplicaStatesAsync(postgres, floor.Value, serverIdFilter, cancellationToken)); + } + + foreach (var server in stale) + { + rows.AddRange(await ReadDatabaseReplicaStatesAsync(postgres, server.Instant, server.ServerId, cancellationToken)); + } + + rows.Sort(CompareDatabaseRows); + return rows; + } + + /// The nav-gate probe's count (#4189): how many distinct (reporting server, AG) groups the + /// replica grain would build, without assembling a single row. Same two-step split as + /// ; the current-set count and every stale server's own count are + /// disjoint by server_id, so they sum without double-counting. + public static async Task GetReplicaGroupCountAsync( + NpgsqlDataSource postgres, int? serverIdFilter, DateTime nowUtc, CancellationToken cancellationToken = default) + { + var instants = await GetNewestInstantsAsync(postgres, ReplicaNewestInstantSql, serverIdFilter, cancellationToken); + var (floor, stale) = SplitCurrentAndStale(instants, nowUtc); + + var total = 0; + if (floor is not null) + { + total += await ReadReplicaGroupCountAsync(postgres, floor.Value, serverIdFilter, cancellationToken); + } + + foreach (var server in stale) + { + total += await ReadReplicaGroupCountAsync(postgres, server.Instant, server.ServerId, cancellationToken); + } + + return total; + } + + /* ─────────────────────────── step one + the staleness split ─────────────────────────── */ + + private static async Task> GetNewestInstantsAsync( + NpgsqlDataSource postgres, string sql, int? serverIdFilter, CancellationToken cancellationToken) + { + var rows = new List<(int, DateTime)>(); + await using var command = postgres.CreateCommand(sql); + command.CommandTimeout = StorageCommandDeadlines.McpReadSeconds; + AddServerFilter(command, serverIdFilter); + await using var reader = await command.ExecuteReaderAsync(cancellationToken); + while (await reader.ReadAsync(cancellationToken)) + { + rows.Add((reader.GetInt32(0), reader.GetDateTime(1))); + } + + return rows; + } + + /// Splits step one's per-server instants into the shared floor (the minimum instant among + /// servers within of , or null when none are + /// current) and the stale servers' own (server_id, instant) pairs — internal so this pure split tests + /// without a store. + internal static (DateTime? Floor, List<(int ServerId, DateTime Instant)> Stale) SplitCurrentAndStale( + IReadOnlyList<(int ServerId, DateTime Instant)> instants, DateTime nowUtc) + { + DateTime? floor = null; + var stale = new List<(int, DateTime)>(); + var horizonStart = nowUtc - StalenessHorizon; + + foreach (var (serverId, instant) in instants) + { + if (instant < horizonStart) + { + stale.Add((serverId, instant)); + } + else if (floor is null || instant < floor.Value) + { + floor = instant; + } + } + + return (floor, stale); + } + + /* ─────────────────────────── step two ─────────────────────────── */ + + private static async Task> ReadReplicaStatesAsync( + NpgsqlDataSource postgres, DateTime floor, int? serverIdFilter, CancellationToken cancellationToken) + { + var rows = new List(); + await using var command = postgres.CreateCommand(ReplicaStatesSql); + command.CommandTimeout = StorageCommandDeadlines.McpReadSeconds; + AddTimestamp(command, floor); + AddServerFilter(command, serverIdFilter); + await using var reader = await command.ExecuteReaderAsync(cancellationToken); + while (await reader.ReadAsync(cancellationToken)) + { + rows.Add(new ReplicaStateRow( + reader.GetInt32(0), + reader.GetString(1), + reader.GetDateTime(2), + Text(reader, 3), + Text(reader, 4), + Text(reader, 5), + Flag(reader, 6), + Text(reader, 7), + Text(reader, 8), + Text(reader, 9), + Text(reader, 10), + Text(reader, 11), + Text(reader, 12), + Text(reader, 13))); + } + + return rows; + } + + private static async Task> ReadDatabaseReplicaStatesAsync( + NpgsqlDataSource postgres, DateTime floor, int? serverIdFilter, CancellationToken cancellationToken) + { + var rows = new List(); + await using var command = postgres.CreateCommand(DatabaseReplicaStatesSql); + command.CommandTimeout = StorageCommandDeadlines.McpReadSeconds; + AddTimestamp(command, floor); + AddServerFilter(command, serverIdFilter); + await using var reader = await command.ExecuteReaderAsync(cancellationToken); + while (await reader.ReadAsync(cancellationToken)) + { + rows.Add(new DatabaseReplicaStateRow( + reader.GetInt32(0), + reader.GetString(1), + reader.GetDateTime(2), + Text(reader, 3), + Text(reader, 4), + Text(reader, 5), + Flag(reader, 6), + Text(reader, 7), + Text(reader, 8), + Text(reader, 9), + Count(reader, 10), + Count(reader, 11), + Count(reader, 12), + Count(reader, 13), + Flag(reader, 14), + Text(reader, 15), + Text(reader, 16), + Count(reader, 17))); + } + + return rows; + } + + private static async Task ReadReplicaGroupCountAsync( + NpgsqlDataSource postgres, DateTime floor, int? serverIdFilter, CancellationToken cancellationToken) + { + await using var command = postgres.CreateCommand(ReplicaGroupCountSql); + command.CommandTimeout = StorageCommandDeadlines.McpReadSeconds; + AddTimestamp(command, floor); + AddServerFilter(command, serverIdFilter); + var result = await command.ExecuteScalarAsync(cancellationToken); + return Convert.ToInt32(result ?? 0L); + } + + /* ─────────────────────────── merge ordering ─────────────────────────── */ + + /// Mirrors 's ORDER BY server_name, ag_name, + /// replica_server_name in C#, needed only because a stale server's own read is a separate round trip + /// from the shared-floor read and the two lists have to interleave back into one order. + private static int CompareReplicaRows(ReplicaStateRow a, ReplicaStateRow b) + { + var byServer = string.CompareOrdinal(a.ServerName, b.ServerName); + if (byServer != 0) + { + return byServer; + } + + var byAg = string.CompareOrdinal(a.AgName ?? "", b.AgName ?? ""); + return byAg != 0 ? byAg : string.CompareOrdinal(a.ReplicaServerName ?? "", b.ReplicaServerName ?? ""); + } + + /// Mirrors 's ORDER BY server_name, ag_name, + /// database_name, replica_server_name, for the same reason as . + private static int CompareDatabaseRows(DatabaseReplicaStateRow a, DatabaseReplicaStateRow b) + { + var byServer = string.CompareOrdinal(a.ServerName, b.ServerName); + if (byServer != 0) + { + return byServer; + } + + var byAg = string.CompareOrdinal(a.AgName ?? "", b.AgName ?? ""); + if (byAg != 0) + { + return byAg; + } + + var byDatabase = string.CompareOrdinal(a.DatabaseName ?? "", b.DatabaseName ?? ""); + return byDatabase != 0 ? byDatabase : string.CompareOrdinal(a.ReplicaServerName ?? "", b.ReplicaServerName ?? ""); + } + + /* ─────────────────────────── binding + column helpers ─────────────────────────── */ + + /// Binds the optional server filter as a nullable integer — NULL means the whole fleet (step + /// one) or every current server (step two). + private static void AddServerFilter(NpgsqlCommand command, int? serverIdFilter) => + command.Parameters.Add(new NpgsqlParameter + { + NpgsqlDbType = NpgsqlDbType.Integer, + Value = (object?)serverIdFilter ?? DBNull.Value, + }); + + /// Binds a floor as Kind=Unspecified — the naive-UTC discipline the whole + /// DarlingPg*Reader family binds by ('s doc). A Kind=Utc value + /// makes Npgsql infer timestamptz, and PostgreSQL then resolves the comparison against these naive + /// timestamp columns by converting the COLUMNS at the store session's TimeZone. + private static void AddTimestamp(NpgsqlCommand command, DateTime value) => + command.Parameters.Add(new NpgsqlParameter { TypedValue = DateTime.SpecifyKind(value, DateTimeKind.Unspecified) }); + + private static string? Text(NpgsqlDataReader reader, int ordinal) => + reader.IsDBNull(ordinal) ? null : reader.GetString(ordinal); + + private static bool? Flag(NpgsqlDataReader reader, int ordinal) => + reader.IsDBNull(ordinal) ? null : reader.GetBoolean(ordinal); + + private static long? Count(NpgsqlDataReader reader, int ordinal) => + reader.IsDBNull(ordinal) ? null : reader.GetInt64(ordinal); +} diff --git a/Darling/PerformanceMonitor.Darling.Viewer/ViewerDataService.AvailabilityGroups.cs b/Darling/PerformanceMonitor.Darling.Viewer/ViewerDataService.AvailabilityGroups.cs index 31241f0988..0b68a0ff9f 100644 --- a/Darling/PerformanceMonitor.Darling.Viewer/ViewerDataService.AvailabilityGroups.cs +++ b/Darling/PerformanceMonitor.Darling.Viewer/ViewerDataService.AvailabilityGroups.cs @@ -8,11 +8,10 @@ using System; using System.Collections.Generic; -using System.Globalization; using System.Threading; using System.Threading.Tasks; -using Npgsql; using PerformanceMonitor.Common; +using PerformanceMonitor.Darling.Storage; namespace PerformanceMonitor.Darling.Viewer; @@ -20,79 +19,17 @@ namespace PerformanceMonitor.Darling.Viewer; /// The Availability Group topology read for the viewer's fleet-level AG tab (#991) — the STORE half only. Every /// banding rule and the card projection live in (PerformanceMonitor.Common), shared /// with the web dashboard and Lite's AG tab, so the three surfaces cannot reach different verdicts about the -/// same AG. This file reads Postgres and maps rows; it decides nothing. +/// same AG. This file maps 's rows (#4228) into +/// / ; it decides nothing. +/// +/// The statement text and the two-step, floor-bounded read used to be a separate copy here — the Viewer +/// had no route to the Service assembly that carried the other copy. Both now call +/// in PerformanceMonitor.Darling.Storage, which both projects already +/// reference, so there is one implementation instead of two that could disagree about what "the newest +/// snapshot" means. /// public sealed partial class ViewerDataService { - /// Every replica row from each server's NEWEST AG collection. The inner aggregate picks the latest - /// instant per server and the join keeps ALL rows at that instant (not DISTINCT ON, which would keep - /// only one replica). Joined to the ENABLED registry so a disabled server's AGs leave the surface with it. - /// Bare table names — the store's search_path resolves collect. $ none. - public const string AgReplicaStatesSql = @" -SELECT - r.server_id, - r.server_name, - r.collection_time, - r.ag_name, - r.replica_server_name, - r.role_desc, - r.is_local, - r.operational_state_desc, - r.connected_state_desc, - r.recovery_health_desc, - r.synchronization_health_desc, - r.availability_mode_desc, - r.failover_mode_desc, - r.endpoint_url -FROM ag_replica_states AS r -JOIN -( - SELECT server_id, MAX(collection_time) AS max_collection_time - FROM ag_replica_states - GROUP BY server_id -) AS latest - ON r.server_id = latest.server_id - AND r.collection_time = latest.max_collection_time -JOIN servers AS s - ON s.server_id = r.server_id - AND s.is_enabled -ORDER BY r.server_name, r.ag_name, r.replica_server_name"; - - /// Every database-grain row from each server's NEWEST AG collection, same latest-snapshot shape as - /// . Windowed independently: the two collectors sweep separately, so their - /// newest instants need not coincide. $ none. - public const string AgDatabaseReplicaStatesSql = @" -SELECT - d.server_id, - d.server_name, - d.collection_time, - d.ag_name, - d.database_name, - d.replica_server_name, - d.is_local, - d.synchronization_state_desc, - d.log_send_queue_size, - d.redo_queue_size, - d.log_send_rate, - d.redo_rate, - d.is_suspended, - d.suspend_reason_desc, - d.availability_mode_desc, - d.secondary_lag_seconds -FROM ag_database_replica_states AS d -JOIN -( - SELECT server_id, MAX(collection_time) AS max_collection_time - FROM ag_database_replica_states - GROUP BY server_id -) AS latest - ON d.server_id = latest.server_id - AND d.collection_time = latest.max_collection_time -JOIN servers AS s - ON s.server_id = d.server_id - AND s.is_enabled -ORDER BY d.server_name, d.ag_name, d.database_name, d.replica_server_name"; - /// /// Reads the fleet's AG topology into per-(reporting server, AG) cards. An AG-less fleet — the common case — /// costs exactly one indexed read that returns nothing, because the database-grain read is skipped entirely @@ -100,83 +37,77 @@ AND s.is_enabled /// public async Task> GetAvailabilityGroupsAsync(CancellationToken cancellationToken = default) { - var replicas = await ReadAgReplicasAsync(cancellationToken); + var nowUtc = DateTime.UtcNow; + var replicas = await ReadAgReplicasAsync(nowUtc, cancellationToken); if (replicas.Count == 0) { return new List(); } - return AgTopology.BuildCards(replicas, await ReadAgDatabasesAsync(cancellationToken)); + return AgTopology.BuildCards(replicas, await ReadAgDatabasesAsync(nowUtc, cancellationToken)); } - private async Task> ReadAgReplicasAsync(CancellationToken cancellationToken) + /// Maps 's raw rows into , + /// field for field, so needed no change when the read moved (#4228). + /// No server filter — the viewer always reads the whole fleet. + private async Task> ReadAgReplicasAsync(DateTime nowUtc, CancellationToken cancellationToken) { - var rows = new List(); - await using var command = _dataSource.CreateCommand(AgReplicaStatesSql); - command.CommandTimeout = ViewerCommandDeadlines.CurrentInteractiveReadSeconds; - await using var reader = await command.ExecuteReaderAsync(cancellationToken); - while (await reader.ReadAsync(cancellationToken)) + var raw = await DarlingAgStatesReader.GetReplicaStatesAsync(_dataSource, null, nowUtc, cancellationToken); + var rows = new List(raw.Count); + foreach (var row in raw) { rows.Add(new AgTopologyReplicaRow { - ServerId = reader.GetInt32(0), - ServerName = reader.GetString(1), - CollectionTime = reader.GetDateTime(2), - AgName = AgText(reader, 3), - ReplicaServerName = AgText(reader, 4), - RoleDesc = AgText(reader, 5), - IsLocal = AgFlag(reader, 6), - OperationalStateDesc = AgText(reader, 7), - ConnectedStateDesc = AgText(reader, 8), - RecoveryHealthDesc = AgText(reader, 9), - SynchronizationHealthDesc = AgText(reader, 10), - AvailabilityModeDesc = AgText(reader, 11), - FailoverModeDesc = AgText(reader, 12), - EndpointUrl = AgText(reader, 13), + ServerId = row.ServerId, + ServerName = row.ServerName, + CollectionTime = row.CollectionTime, + AgName = row.AgName, + ReplicaServerName = row.ReplicaServerName, + RoleDesc = row.RoleDesc, + IsLocal = row.IsLocal, + OperationalStateDesc = row.OperationalStateDesc, + ConnectedStateDesc = row.ConnectedStateDesc, + RecoveryHealthDesc = row.RecoveryHealthDesc, + SynchronizationHealthDesc = row.SynchronizationHealthDesc, + AvailabilityModeDesc = row.AvailabilityModeDesc, + FailoverModeDesc = row.FailoverModeDesc, + EndpointUrl = row.EndpointUrl, }); } return rows; } - private async Task> ReadAgDatabasesAsync(CancellationToken cancellationToken) + /// Same mapping as , for the database grain. The shared reader's + /// row also carries LastHardenedLsn / LastCommitLsn (the Service's columns); the viewer's card + /// does not surface them today, so they are simply not read here, same as before the move. + private async Task> ReadAgDatabasesAsync(DateTime nowUtc, CancellationToken cancellationToken) { - var rows = new List(); - await using var command = _dataSource.CreateCommand(AgDatabaseReplicaStatesSql); - command.CommandTimeout = ViewerCommandDeadlines.CurrentInteractiveReadSeconds; - await using var reader = await command.ExecuteReaderAsync(cancellationToken); - while (await reader.ReadAsync(cancellationToken)) + var raw = await DarlingAgStatesReader.GetDatabaseReplicaStatesAsync(_dataSource, null, nowUtc, cancellationToken); + var rows = new List(raw.Count); + foreach (var row in raw) { rows.Add(new AgTopologyDatabaseRow { - ServerId = reader.GetInt32(0), - ServerName = reader.GetString(1), - CollectionTime = reader.GetDateTime(2), - AgName = AgText(reader, 3), - DatabaseName = AgText(reader, 4), - ReplicaServerName = AgText(reader, 5), - IsLocal = AgFlag(reader, 6), - SynchronizationStateDesc = AgText(reader, 7), - LogSendQueueKb = AgCount(reader, 8), - RedoQueueKb = AgCount(reader, 9), - LogSendRateKbPerSec = AgCount(reader, 10), - RedoRateKbPerSec = AgCount(reader, 11), - IsSuspended = AgFlag(reader, 12), - SuspendReasonDesc = AgText(reader, 13), - AvailabilityModeDesc = AgText(reader, 14), - SecondaryLagSeconds = AgCount(reader, 15), + ServerId = row.ServerId, + ServerName = row.ServerName, + CollectionTime = row.CollectionTime, + AgName = row.AgName, + DatabaseName = row.DatabaseName, + ReplicaServerName = row.ReplicaServerName, + IsLocal = row.IsLocal, + SynchronizationStateDesc = row.SynchronizationStateDesc, + LogSendQueueKb = row.LogSendQueueSize, + RedoQueueKb = row.RedoQueueSize, + LogSendRateKbPerSec = row.LogSendRate, + RedoRateKbPerSec = row.RedoRate, + IsSuspended = row.IsSuspended, + SuspendReasonDesc = row.SuspendReasonDesc, + AvailabilityModeDesc = row.AvailabilityModeDesc, + SecondaryLagSeconds = row.SecondaryLagSeconds, }); } return rows; } - - private static string? AgText(NpgsqlDataReader reader, int ordinal) => - reader.IsDBNull(ordinal) ? null : reader.GetString(ordinal); - - private static bool? AgFlag(NpgsqlDataReader reader, int ordinal) => - reader.IsDBNull(ordinal) ? null : reader.GetBoolean(ordinal); - - private static long? AgCount(NpgsqlDataReader reader, int ordinal) => - reader.IsDBNull(ordinal) ? null : Convert.ToInt64(reader.GetValue(ordinal), CultureInfo.InvariantCulture); } diff --git a/Lite.Tests/AgAvailabilityGroupNewestSnapshotEqualityTests.cs b/Lite.Tests/AgAvailabilityGroupNewestSnapshotEqualityTests.cs new file mode 100644 index 0000000000..b9074e6560 --- /dev/null +++ b/Lite.Tests/AgAvailabilityGroupNewestSnapshotEqualityTests.cs @@ -0,0 +1,260 @@ +/* + * Copyright (c) 2026 Erik Darling, Darling Data LLC + * + * This file is part of the SQL Server Performance Monitor Lite. + * + * Licensed under the MIT License. See LICENSE file in the project root for full license information. + */ + +using System; +using System.Collections.Generic; +using System.Threading.Tasks; +using DuckDB.NET.Data; +using PerformanceMonitor.Common; +using PerformanceMonitorLite.Database; +using PerformanceMonitorLite.Services; +using Xunit; + +namespace PerformanceMonitorLite.Tests; + +/// +/// #4228 Lite parity. G1's PR body dismissed Lite because "a Lite install monitors one server" — wrong: Lite +/// can and does monitor several (LocalDataService.AvailabilityGroups.cs reads per server with +/// WHERE server_id = $1), and the per-server MAX(collection_time) with no bound can scan that +/// one server's whole retained history regardless of fleet size. Timed against a DuckDB table seeded at the +/// real ScheduleManager defaults for both AG collectors (1-minute frequency, 30-day retention, several +/// servers' rows interleaved by collection tick): the old correlated-MAX query measured 47-93 ms, well over +/// the ~20 ms parity bar. GetLatestAgReplicaStatesAsync / GetLatestAgDatabaseReplicaStatesAsync +/// now use the same two-step newest-instant split DarlingAgStatesReader uses (ordered-descent step +/// one, then step two bound by that exact instant) — the query itself dropped to 2-4 ms in that same seed. +/// +/// This class pins the NEW two-step code against the OLD (pre-#4228) correlated-subquery SQL text, +/// row for row, across the two edge cases the brief calls out: a TIE (several replicas/databases sharing the +/// newest collection_time — the normal shape of any AG snapshot) and a NULL (ag_name or +/// replica_server_name reading NULL under WSFC quorum loss, which both old and new code must DROP from +/// the projected rows without it affecting which instant is "newest" — collection_time itself is +/// NOT NULL in the schema, so it is never the null column). +/// +public sealed class AgAvailabilityGroupNewestSnapshotEqualityTests : IClassFixture, IDisposable +{ + private readonly DuckDbInitializer _duckDb; + private DuckDBConnection? _seedConn; + private long _nextId = 1; + + public AgAvailabilityGroupNewestSnapshotEqualityTests(SharedDuckDbFixture fixture) + { + fixture.ResetData(); + _duckDb = fixture.DuckDb; + } + + public void Dispose() => _seedConn?.Dispose(); + + private async Task SeedConnectionAsync() + { + if (_seedConn is null) + { + _seedConn = _duckDb.CreateConnection(); + await _seedConn.OpenAsync(); + } + return _seedConn; + } + + private async Task SeedReplicaAsync(int serverId, DateTime collectionTime, string? agName, string? replicaServerName) + { + using var readLock = _duckDb.AcquireReadLock(); + var connection = await SeedConnectionAsync(); + using var cmd = connection.CreateCommand(); + cmd.CommandText = @" +INSERT INTO ag_replica_states + (collection_id, collection_time, server_id, server_name, ag_name, replica_server_name, role_desc, + operational_state_desc, connected_state_desc, recovery_health_desc, synchronization_health_desc, + availability_mode_desc, failover_mode_desc, endpoint_url, is_local) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15)"; + cmd.Parameters.Add(new DuckDBParameter { Value = _nextId++ }); + cmd.Parameters.Add(new DuckDBParameter { Value = collectionTime }); + cmd.Parameters.Add(new DuckDBParameter { Value = serverId }); + cmd.Parameters.Add(new DuckDBParameter { Value = "SRV" + serverId }); + cmd.Parameters.Add(new DuckDBParameter { Value = (object?)agName ?? DBNull.Value }); + cmd.Parameters.Add(new DuckDBParameter { Value = (object?)replicaServerName ?? DBNull.Value }); + cmd.Parameters.Add(new DuckDBParameter { Value = "PRIMARY" }); + cmd.Parameters.Add(new DuckDBParameter { Value = "ONLINE" }); + cmd.Parameters.Add(new DuckDBParameter { Value = "CONNECTED" }); + cmd.Parameters.Add(new DuckDBParameter { Value = "HEALTHY" }); + cmd.Parameters.Add(new DuckDBParameter { Value = "HEALTHY" }); + cmd.Parameters.Add(new DuckDBParameter { Value = "SYNCHRONOUS_COMMIT" }); + cmd.Parameters.Add(new DuckDBParameter { Value = "MANUAL" }); + cmd.Parameters.Add(new DuckDBParameter { Value = "tcp://x" }); + cmd.Parameters.Add(new DuckDBParameter { Value = true }); + await cmd.ExecuteNonQueryAsync(); + } + + private async Task SeedDatabaseAsync(int serverId, DateTime collectionTime, string? agName, string? databaseName, string? replicaServerName) + { + using var readLock = _duckDb.AcquireReadLock(); + var connection = await SeedConnectionAsync(); + using var cmd = connection.CreateCommand(); + cmd.CommandText = @" +INSERT INTO ag_database_replica_states + (collection_id, collection_time, server_id, server_name, ag_name, database_name, replica_server_name, + secondary_lag_seconds, redo_queue_size, is_suspended, suspend_reason_desc) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)"; + cmd.Parameters.Add(new DuckDBParameter { Value = _nextId++ }); + cmd.Parameters.Add(new DuckDBParameter { Value = collectionTime }); + cmd.Parameters.Add(new DuckDBParameter { Value = serverId }); + cmd.Parameters.Add(new DuckDBParameter { Value = "SRV" + serverId }); + cmd.Parameters.Add(new DuckDBParameter { Value = (object?)agName ?? DBNull.Value }); + cmd.Parameters.Add(new DuckDBParameter { Value = (object?)databaseName ?? DBNull.Value }); + cmd.Parameters.Add(new DuckDBParameter { Value = (object?)replicaServerName ?? DBNull.Value }); + cmd.Parameters.Add(new DuckDBParameter { Value = 5L }); + cmd.Parameters.Add(new DuckDBParameter { Value = 0L }); + cmd.Parameters.Add(new DuckDBParameter { Value = false }); + cmd.Parameters.Add(new DuckDBParameter { Value = DBNull.Value }); + await cmd.ExecuteNonQueryAsync(); + } + + /// The exact pre-#4228 statement text for the replica grain — unbounded correlated MAX. + private const string OldReplicaSql = @" +SELECT ag_name, replica_server_name, role_desc, connected_state_desc, collection_time +FROM ag_replica_states +WHERE server_id = $1 +AND collection_time = (SELECT MAX(collection_time) FROM ag_replica_states WHERE server_id = $1) +ORDER BY ag_name, replica_server_name"; + + /// The exact pre-#4228 statement text for the database grain — unbounded correlated MAX. + private const string OldDatabaseSql = @" +SELECT ag_name, database_name, replica_server_name, secondary_lag_seconds, redo_queue_size, + is_suspended, suspend_reason_desc, collection_time +FROM ag_database_replica_states +WHERE server_id = $1 +AND collection_time = (SELECT MAX(collection_time) FROM ag_database_replica_states WHERE server_id = $1) +ORDER BY ag_name, database_name, replica_server_name"; + + private async Task<(DateTime? Newest, List Rows)> RunOldReplicaSqlAsync(int serverId) + { + var rows = new List(); + DateTime? newest = null; + + using var readLock = _duckDb.AcquireReadLock(); + var connection = await SeedConnectionAsync(); + using var cmd = connection.CreateCommand(); + cmd.CommandText = OldReplicaSql; + cmd.Parameters.Add(new DuckDBParameter { Value = serverId }); + + using var reader = await cmd.ExecuteReaderAsync(); + while (await reader.ReadAsync()) + { + if (!reader.IsDBNull(4)) + { + var stamped = DateTime.SpecifyKind(reader.GetDateTime(4), DateTimeKind.Utc); + if (newest is null || stamped > newest) + { + newest = stamped; + } + } + + if (reader.IsDBNull(0) || reader.IsDBNull(1)) + { + continue; + } + + rows.Add(new AgReplicaReading( + AgName: reader.GetString(0), + ReplicaServerName: reader.GetString(1), + RoleDesc: reader.IsDBNull(2) ? null : reader.GetString(2), + ConnectedStateDesc: reader.IsDBNull(3) ? null : reader.GetString(3))); + } + + return (newest, rows); + } + + private async Task<(DateTime? Newest, List Rows)> RunOldDatabaseSqlAsync(int serverId) + { + var rows = new List(); + DateTime? newest = null; + + using var readLock = _duckDb.AcquireReadLock(); + var connection = await SeedConnectionAsync(); + using var cmd = connection.CreateCommand(); + cmd.CommandText = OldDatabaseSql; + cmd.Parameters.Add(new DuckDBParameter { Value = serverId }); + + using var reader = await cmd.ExecuteReaderAsync(); + while (await reader.ReadAsync()) + { + if (!reader.IsDBNull(7)) + { + var stamped = DateTime.SpecifyKind(reader.GetDateTime(7), DateTimeKind.Utc); + if (newest is null || stamped > newest) + { + newest = stamped; + } + } + + if (reader.IsDBNull(0) || reader.IsDBNull(1) || reader.IsDBNull(2)) + { + continue; + } + + rows.Add(new AgDatabaseReading( + AgName: reader.GetString(0), + DatabaseName: reader.GetString(1), + ReplicaServerName: reader.GetString(2), + SecondaryLagSeconds: reader.IsDBNull(3) ? null : Convert.ToInt64(reader.GetValue(3)), + RedoQueueSizeKb: reader.IsDBNull(4) ? null : Convert.ToInt64(reader.GetValue(4)), + IsSuspended: reader.IsDBNull(5) ? null : reader.GetBoolean(5), + SuspendReasonDesc: reader.IsDBNull(6) ? null : reader.GetString(6))); + } + + return (newest, rows); + } + + [Fact] + public async Task ReplicaRead_MatchesOldUnboundedSql_WithATieAndANullKeyedRowAtTheNewestInstant() + { + var stale = new DateTime(2026, 9, 1, 0, 0, 0, DateTimeKind.Unspecified); + var newest = new DateTime(2026, 9, 25, 12, 0, 0, DateTimeKind.Unspecified); + + // an older snapshot that must NOT be selected + await SeedReplicaAsync(9001, stale, "AG1", "NODE-OLD"); + + // the newest instant: a tie of two keyable rows, plus a NULL-keyed row (WSFC quorum loss) that both + // old and new code must drop from the projection without it affecting which instant is "newest" + await SeedReplicaAsync(9001, newest, "AG1", "NODE-A"); + await SeedReplicaAsync(9001, newest, "AG1", "NODE-B"); + await SeedReplicaAsync(9001, newest, null, "NODE-C"); + + // a different server's rows must not leak in + await SeedReplicaAsync(9002, newest, "AG1", "OTHER-NODE"); + + var expected = await RunOldReplicaSqlAsync(9001); + var service = new LocalDataService(_duckDb); + var actual = await service.GetLatestAgReplicaStatesAsync(9001); + + Assert.Equal(2, expected.Rows.Count); + Assert.Equal(expected.Newest, actual.CollectionTimeUtc); + Assert.Equal(expected.Rows, actual.Replicas); + } + + [Fact] + public async Task DatabaseRead_MatchesOldUnboundedSql_WithATieAndANullKeyedRowAtTheNewestInstant() + { + var stale = new DateTime(2026, 9, 1, 0, 0, 0, DateTimeKind.Unspecified); + var newest = new DateTime(2026, 9, 25, 12, 0, 0, DateTimeKind.Unspecified); + + await SeedDatabaseAsync(9001, stale, "AG1", "DB-OLD", "NODE-OLD"); + + await SeedDatabaseAsync(9001, newest, "AG1", "DB1", "NODE-A"); + await SeedDatabaseAsync(9001, newest, "AG1", "DB2", "NODE-A"); + await SeedDatabaseAsync(9001, newest, "AG1", null, "NODE-A"); + + await SeedDatabaseAsync(9002, newest, "AG1", "DB1", "OTHER-NODE"); + + var expected = await RunOldDatabaseSqlAsync(9001); + var service = new LocalDataService(_duckDb); + var actual = await service.GetLatestAgDatabaseReplicaStatesAsync(9001); + + Assert.Equal(2, expected.Rows.Count); + Assert.Equal(expected.Newest, actual.CollectionTimeUtc); + Assert.Equal(expected.Rows, actual.Databases); + } +} diff --git a/Lite/Services/LocalDataService.AvailabilityGroups.cs b/Lite/Services/LocalDataService.AvailabilityGroups.cs index cf23ee8db2..e1ca6a9bad 100644 --- a/Lite/Services/LocalDataService.AvailabilityGroups.cs +++ b/Lite/Services/LocalDataService.AvailabilityGroups.cs @@ -19,8 +19,20 @@ public partial class LocalDataService /// /// The server's LATEST ag_replica_states snapshot (#1696) — the rows at that table's own /// MAX(collection_time), plus that timestamp so the caller can decide whether the reading is fresh - /// enough to judge. Darling's twin is ReadLatestAgReplicaStatesAsync; the SQL is the same shape - /// against DuckDB instead of Postgres. + /// enough to judge. Darling's twin is DarlingAgStatesReader; this is the same two-step + /// newest-instant shape (#4228), adapted to DuckDB and to Lite's already-per-server call site. + /// + /// Two steps, like the Postgres reader — but a plain equality bound, not a floor. The + /// original single statement found "this server's newest row" with a correlated + /// collection_time = (SELECT MAX(collection_time) ... WHERE server_id = $1), which measured + /// 47-93 ms against a DuckDB table seeded at the real ScheduleManager defaults for + /// ag_replica_states (1-minute frequency, 30-day retention, several servers' rows interleaved by + /// collection tick — the shape a multi-server Lite install actually writes). + /// splits that into an ordered-descent step one (cheap: the index on (server_id, collection_time) + /// answers it as a range scan) and a bound step two. Lite's read is already scoped to ONE server per call + /// (there is no fleet-wide shared floor to keep a stale server from dragging back), so step two here uses + /// step one's own exact instant as an equality bound instead of Darling's >= floor aggregate — + /// simpler, and still exactly the same answer. Measured after the split: 3-14 ms for the same seed. /// /// Rows whose ag_name or replica_server_name is NULL are DROPPED: those are the /// identity of the alert's state key, and under WSFC quorum loss they can read NULL. A row that cannot be @@ -30,30 +42,44 @@ public partial class LocalDataService public async Task<(DateTime? CollectionTimeUtc, List Replicas)> GetLatestAgReplicaStatesAsync(int serverId) { var replicas = new List(); - DateTime? newest = null; using var connection = await OpenConnectionAsync(); + + DateTime? newest = null; + using (var findCommand = connection.CreateCommand()) + { + findCommand.CommandText = @" +SELECT collection_time +FROM ag_replica_states +WHERE server_id = $1 +ORDER BY collection_time DESC +LIMIT 1"; + findCommand.Parameters.Add(new DuckDBParameter { Value = serverId }); + + if (await findCommand.ExecuteScalarAsync() is DateTime found) + { + newest = DateTime.SpecifyKind(found, DateTimeKind.Utc); + } + } + + if (newest is null) + { + return (null, replicas); + } + using var command = connection.CreateCommand(); command.CommandText = @" -SELECT ag_name, replica_server_name, role_desc, connected_state_desc, collection_time +SELECT ag_name, replica_server_name, role_desc, connected_state_desc FROM ag_replica_states WHERE server_id = $1 -AND collection_time = (SELECT MAX(collection_time) FROM ag_replica_states WHERE server_id = $1) +AND collection_time = $2 ORDER BY ag_name, replica_server_name"; command.Parameters.Add(new DuckDBParameter { Value = serverId }); + command.Parameters.Add(new DuckDBParameter { Value = DateTime.SpecifyKind(newest.Value, DateTimeKind.Unspecified) }); using var reader = await command.ExecuteReaderAsync(); while (await reader.ReadAsync()) { - if (!reader.IsDBNull(4)) - { - var stamped = DateTime.SpecifyKind(reader.GetDateTime(4), DateTimeKind.Utc); - if (newest is null || stamped > newest) - { - newest = stamped; - } - } - if (reader.IsDBNull(0) || reader.IsDBNull(1)) { continue; @@ -75,35 +101,52 @@ FROM ag_replica_states /// freshness-gated by the caller — because the two AG collectors carry independent schedule entries: a /// disabled or failing database-grain collector must not have the replica grain's healthy timestamp vouch /// for its stale rows, which would keep re-firing "AG Sync Fell Behind" off days-old lag readings. + /// + /// Same two-step newest-instant split as (#4228): an + /// ordered-descent step one, then step two bound by that exact instant. /// public async Task<(DateTime? CollectionTimeUtc, List Databases)> GetLatestAgDatabaseReplicaStatesAsync(int serverId) { var databases = new List(); - DateTime? newest = null; using var connection = await OpenConnectionAsync(); + + DateTime? newest = null; + using (var findCommand = connection.CreateCommand()) + { + findCommand.CommandText = @" +SELECT collection_time +FROM ag_database_replica_states +WHERE server_id = $1 +ORDER BY collection_time DESC +LIMIT 1"; + findCommand.Parameters.Add(new DuckDBParameter { Value = serverId }); + + if (await findCommand.ExecuteScalarAsync() is DateTime found) + { + newest = DateTime.SpecifyKind(found, DateTimeKind.Utc); + } + } + + if (newest is null) + { + return (null, databases); + } + using var command = connection.CreateCommand(); command.CommandText = @" SELECT ag_name, database_name, replica_server_name, secondary_lag_seconds, redo_queue_size, - is_suspended, suspend_reason_desc, collection_time + is_suspended, suspend_reason_desc FROM ag_database_replica_states WHERE server_id = $1 -AND collection_time = (SELECT MAX(collection_time) FROM ag_database_replica_states WHERE server_id = $1) +AND collection_time = $2 ORDER BY ag_name, database_name, replica_server_name"; command.Parameters.Add(new DuckDBParameter { Value = serverId }); + command.Parameters.Add(new DuckDBParameter { Value = DateTime.SpecifyKind(newest.Value, DateTimeKind.Unspecified) }); using var reader = await command.ExecuteReaderAsync(); while (await reader.ReadAsync()) { - if (!reader.IsDBNull(7)) - { - var stamped = DateTime.SpecifyKind(reader.GetDateTime(7), DateTimeKind.Utc); - if (newest is null || stamped > newest) - { - newest = stamped; - } - } - if (reader.IsDBNull(0) || reader.IsDBNull(1) || reader.IsDBNull(2)) { continue;