From 1c443a9d29f7085884b6aeebb4ba718df8e8089c Mon Sep 17 00:00:00 2001 From: Erik Darling <2136037+erikdarlingdata@users.noreply.github.com> Date: Sat, 19 Sep 2026 10:07:03 -0400 Subject: [PATCH] get_fleet_overview: one collection-health scan per minute per host, however many overview calls race (#3735) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The 7-day collection-health rollup inside DarlingFleetReader.GetFleetOverviewAsync is window-independent — FleetCollectionHealthSql's $1 is always now - 7 days, and hours_back bounds only the blocking and deadlock reads — so a caller racing three overview calls with three windows had the store run three identical copies of one 7-day aggregate concurrently. Photographed 2026-09-19 13:36Z on the largest production store: three copies, two parallel workers each, nine backends on collection_log in pure I/O waits inside the collectors' flush band, the slowest crossing the mcp role's 15 s statement_timeout. Solo the plan runs in 680 ms. CollectionHealthMemo single-flights that read per NpgsqlDataSource: concurrent callers share one in-flight scan, a completed scan under 60 s old (the fastest collector cadence — nothing the rollup bands can move faster) is served from memory, a failed scan is never memoized, and the shared scan runs on CancellationToken.None bounded by the 20 s command deadline so one caller's departure releases only that caller. FleetOverviewResult gains a trailing collection_health_age_seconds (0 on a fresh scan) on both JSON surfaces. The SQL text, the other twelve reads, McpCommandDeadlines and hours_back semantics are unchanged. Tests: the memo through its Func seam (one scan for five racing callers; served with its age under a minute and re-read at 60; failure not memoized and not evicting the last good reading; the first caller's cancellation releasing only that caller; abandoned scans still memoized; one memo per data source; the age riding BuildRollup and serializing trailing; the reader reaching the statement only through the memo) plus a gated live test racing three real GetFleetOverviewAsync calls against one store and reading ScansStarted == 1. --- .../FleetCollectionHealthMemoTests.cs | 423 ++++++++++++++++++ .../Mcp/DarlingFleetReader.cs | 245 +++++++++- .../Mcp/DarlingMcpFleetTools.cs | 10 +- .../Mcp/DarlingMcpInstructions.cs | 2 +- 4 files changed, 675 insertions(+), 5 deletions(-) create mode 100644 Darling/Darling.Tests/FleetCollectionHealthMemoTests.cs diff --git a/Darling/Darling.Tests/FleetCollectionHealthMemoTests.cs b/Darling/Darling.Tests/FleetCollectionHealthMemoTests.cs new file mode 100644 index 0000000000..4a9227e266 --- /dev/null +++ b/Darling/Darling.Tests/FleetCollectionHealthMemoTests.cs @@ -0,0 +1,423 @@ +/* + * 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.Text.Json; +using System.Text.RegularExpressions; +using System.Threading; +using System.Threading.Tasks; +using Npgsql; +using PerformanceMonitor.Darling.Service.Mcp; +using PerformanceMonitor.Darling.Storage; +using Xunit; + +namespace Darling.Tests; + +/// +/// The single-flight memo over get_fleet_overview's 7-day collection-health rollup (#3735). The +/// production photo: one caller raced three overview calls with three hours_back values, and because +/// that rollup does not depend on the window, the store ran three identical copies of one 7-day aggregate +/// concurrently inside the collectors' flush band and the slowest crossed the role's 15 s +/// statement_timeout. These pins hold the four promises the fix makes — one scan however many callers +/// race, a minute of memory, a failed scan never memoized, and the age published beside the payload — and the +/// one it makes about cancellation, which is the riskiest line in the change: a caller's token releases that +/// caller and nobody else. +/// +/// The memo is exercised through its Func seam with a counting scan rather than through a mocked +/// Npgsql, and the seam is the SAME object the reader routes through, so the count here IS the number of +/// statements the store would have seen. The gated live test at the bottom closes the loop against a real +/// store with three racing GetFleetOverviewAsync calls. +/// +public sealed class FleetCollectionHealthMemoTests +{ + private static readonly DateTime T0 = new(2026, 9, 19, 13, 36, 3, DateTimeKind.Utc); + + /// A scan that counts how often it was invoked, records the token it was handed, and completes + /// only when the test says so — so N callers can be parked on it before it produces anything, which is + /// the only way to prove they shared it rather than merely ran in sequence. + private sealed class CountingScan + { + private int _calls; + public TaskCompletionSource> Gate { get; private set; } = + new(TaskCreationOptions.RunContinuationsAsynchronously); + public List TokensSeen { get; } = new(); + public int Calls => Volatile.Read(ref _calls); + + public Task> Run(DateTime now, CancellationToken token) + { + Interlocked.Increment(ref _calls); + lock (TokensSeen) + { + TokensSeen.Add(token); + } + + return Gate.Task; + } + + public void Release(int healthyForServerOne = 1) + { + var gate = Gate; + Gate = new(TaskCreationOptions.RunContinuationsAsynchronously); + gate.SetResult(new Dictionary + { + [1] = new(healthyForServerOne, 0, healthyForServerOne), + }); + } + + public void Fail(Exception ex) + { + var gate = Gate; + Gate = new(TaskCreationOptions.RunContinuationsAsynchronously); + gate.SetException(ex); + } + } + + /// (a) Five concurrent callers, all parked before the scan produces anything, cost the store ONE + /// statement — and every one of them gets the same reading at age 0, because the scan was cut at the + /// instant they all asked for. + [Fact] + public async Task ConcurrentCallers_ShareOneScan() + { + var memo = new DarlingFleetReader.CollectionHealthMemo(); + var scan = new CountingScan(); + + var callers = Enumerable.Range(0, 5) + .Select(_ => memo.GetAsync(T0, scan.Run, CancellationToken.None)) + .ToArray(); + + /* Every caller is in flight and the statement has run exactly once before anything completes. */ + Assert.All(callers, c => Assert.False(c.IsCompleted)); + Assert.Equal(1, scan.Calls); + Assert.Equal(1, memo.ScansStarted); + + scan.Release(); + var results = await Task.WhenAll(callers); + + Assert.Equal(1, scan.Calls); + Assert.All(results, r => Assert.Equal(0, r.AgeSeconds)); + Assert.All(results, r => Assert.Same(results[0].Counts, r.Counts)); + Assert.Equal(1, results[0].Counts[1].Healthy); + } + + /// (b) and (d) together, because they are one clock: a reading is served for 60 seconds with its + /// age beside it, and re-read at 60 — the lifetime is the fastest collector cadence, asserted as a value + /// so a change to it has to change this line too. + [Fact] + public async Task AReadingUnderAMinuteOld_IsServedWithItsAge_AndReReadAtSixty() + { + Assert.Equal(TimeSpan.FromSeconds(60), DarlingFleetReader.CollectionHealthMemoLifetime); + + var memo = new DarlingFleetReader.CollectionHealthMemo(); + var scan = new CountingScan(); + + var first = memo.GetAsync(T0, scan.Run, CancellationToken.None); + scan.Release(healthyForServerOne: 7); + var fresh = await first; + Assert.Equal(0, fresh.AgeSeconds); + Assert.Equal(7, fresh.Counts[1].Healthy); + + /* 30 s later: a hit, age 30, no statement. 59 s later: still a hit. */ + var hit = await memo.GetAsync(T0.AddSeconds(30), scan.Run, CancellationToken.None); + Assert.Equal(30, hit.AgeSeconds); + Assert.Same(fresh.Counts, hit.Counts); + var lastHit = await memo.GetAsync(T0.AddSeconds(59.9), scan.Run, CancellationToken.None); + Assert.Equal(59, lastHit.AgeSeconds); + Assert.Equal(1, scan.Calls); + + /* At 60 s the reading is re-read: a second statement, a new dictionary, age back to 0. */ + var expiredCall = memo.GetAsync(T0.AddSeconds(60), scan.Run, CancellationToken.None); + Assert.Equal(2, scan.Calls); + scan.Release(healthyForServerOne: 8); + var reread = await expiredCall; + Assert.Equal(0, reread.AgeSeconds); + Assert.Equal(8, reread.Counts[1].Healthy); + Assert.NotSame(fresh.Counts, reread.Counts); + Assert.Equal(2, memo.ScansStarted); + } + + /// A caller whose now reads BEHIND the memo's — two hosts' clocks, or a test's fixed + /// instant — is not holding a reading from the future; it is holding the freshest one there is, and the + /// age says 0 rather than going negative or forcing a pointless re-read. + [Fact] + public async Task ACallerWhoseClockIsBehindTheMemo_IsServedAtAgeZero() + { + var memo = new DarlingFleetReader.CollectionHealthMemo(); + var scan = new CountingScan(); + + var first = memo.GetAsync(T0, scan.Run, CancellationToken.None); + scan.Release(); + await first; + + var behind = await memo.GetAsync(T0.AddSeconds(-5), scan.Run, CancellationToken.None); + Assert.Equal(0, behind.AgeSeconds); + Assert.Equal(1, scan.Calls); + } + + /// (c) A failed scan is not memoized: every waiter present sees the failure — the ORIGINAL + /// exception, not a wrapper — and the very next caller at the same instant runs a fresh statement rather + /// than being handed the failure again or an empty reading. + [Fact] + public async Task AFailedScan_IsNotMemoized_TheNextCallReRuns() + { + var memo = new DarlingFleetReader.CollectionHealthMemo(); + var scan = new CountingScan(); + + var a = memo.GetAsync(T0, scan.Run, CancellationToken.None); + var b = memo.GetAsync(T0, scan.Run, CancellationToken.None); + Assert.Equal(1, scan.Calls); + + scan.Fail(new InvalidOperationException("57014: canceling statement due to statement timeout")); + + var exA = await Assert.ThrowsAsync(() => a); + var exB = await Assert.ThrowsAsync(() => b); + Assert.Contains("57014", exA.Message, StringComparison.Ordinal); + Assert.Same(exA, exB); + + /* Nothing was memoized: the next caller, same instant, starts statement #2 and gets its reading. */ + var retry = memo.GetAsync(T0, scan.Run, CancellationToken.None); + Assert.Equal(2, scan.Calls); + scan.Release(healthyForServerOne: 3); + var reading = await retry; + Assert.Equal(0, reading.AgeSeconds); + Assert.Equal(3, reading.Counts[1].Healthy); + } + + /// A failure does not evict a GOOD reading either: while the last successful scan is under a + /// minute old it keeps being served, and only a caller past the minute pays for the re-read. Pinned + /// because the tempting simplification — "on failure clear everything" — would turn one timed-out + /// statement into a burst of retries from every caller of the next minute. + [Fact] + public async Task AFailedReRead_DoesNotEvictTheLastGoodReading() + { + var memo = new DarlingFleetReader.CollectionHealthMemo(); + var scan = new CountingScan(); + + var first = memo.GetAsync(T0, scan.Run, CancellationToken.None); + scan.Release(healthyForServerOne: 5); + var good = await first; + + /* 61 s on: the re-read fails. Its caller sees the failure. */ + var failing = memo.GetAsync(T0.AddSeconds(61), scan.Run, CancellationToken.None); + Assert.Equal(2, scan.Calls); + scan.Fail(new TimeoutException("simulated")); + await Assert.ThrowsAsync(() => failing); + + /* A caller whose now is still inside the good reading's minute gets the good reading. */ + var stillGood = await memo.GetAsync(T0.AddSeconds(45), scan.Run, CancellationToken.None); + Assert.Same(good.Counts, stillGood.Counts); + Assert.Equal(45, stillGood.AgeSeconds); + Assert.Equal(2, scan.Calls); + + /* And a caller past it starts statement #3 rather than inheriting the failure. */ + var third = memo.GetAsync(T0.AddSeconds(62), scan.Run, CancellationToken.None); + Assert.Equal(3, scan.Calls); + scan.Release(healthyForServerOne: 6); + Assert.Equal(6, (await third).Counts[1].Healthy); + } + + /// + /// The riskiest line in the change, stated as behaviour: when the FIRST caller cancels while two others + /// wait, that caller is released at once with a cancellation, the statement keeps running for the two, + /// and its result is memoized for everyone who arrives inside the next minute. The scan itself never sees + /// a cancellable token — it runs bounded by the command deadline and the role's statement_timeout, not by + /// any one caller's patience — so a departing browser tab cannot turn one scan into cancel-and-restart + /// churn. + /// + [Fact] + public async Task TheFirstCallersCancellation_ReleasesOnlyThatCaller_AndTheScanCompletesForTheRest() + { + var memo = new DarlingFleetReader.CollectionHealthMemo(); + var scan = new CountingScan(); + + using var firstCallersToken = new CancellationTokenSource(); + var first = memo.GetAsync(T0, scan.Run, firstCallersToken.Token); + var second = memo.GetAsync(T0, scan.Run, CancellationToken.None); + var third = memo.GetAsync(T0, scan.Run, CancellationToken.None); + Assert.Equal(1, scan.Calls); + + /* The statement was handed a token that cannot cancel — the first caller's never reached it. */ + Assert.Single(scan.TokensSeen); + Assert.False(scan.TokensSeen[0].CanBeCanceled); + + firstCallersToken.Cancel(); + await Assert.ThrowsAnyAsync(() => first); + + /* The other two are still waiting on a scan that is still running — nothing was torn down. */ + Assert.False(second.IsCompleted); + Assert.False(third.IsCompleted); + Assert.Equal(1, scan.Calls); + + scan.Release(healthyForServerOne: 9); + var results = await Task.WhenAll(second, third); + Assert.All(results, r => Assert.Equal(9, r.Counts[1].Healthy)); + Assert.All(results, r => Assert.Equal(0, r.AgeSeconds)); + + /* And the departed caller's cancellation cost nobody a statement: a fourth caller 20 s on is a hit. */ + var fourth = await memo.GetAsync(T0.AddSeconds(20), scan.Run, CancellationToken.None); + Assert.Equal(20, fourth.AgeSeconds); + Assert.Same(results[0].Counts, fourth.Counts); + Assert.Equal(1, scan.Calls); + Assert.Equal(1, memo.ScansStarted); + } + + /// The other half of that promise: a scan EVERY waiter abandoned still completes and is still + /// memoized — the statement was already paid for, and the next tick should not pay for it again. + [Fact] + public async Task AScanEveryWaiterAbandoned_StillCompletes_AndIsStillMemoized() + { + var memo = new DarlingFleetReader.CollectionHealthMemo(); + var scan = new CountingScan(); + + using var cts = new CancellationTokenSource(); + var a = memo.GetAsync(T0, scan.Run, cts.Token); + var b = memo.GetAsync(T0, scan.Run, cts.Token); + cts.Cancel(); + await Assert.ThrowsAnyAsync(() => a); + await Assert.ThrowsAnyAsync(() => b); + + scan.Release(healthyForServerOne: 4); + + /* The orphaned scan's result landed: a later caller inside the minute is a hit on it. The memo's + write happens on the scan's continuation, so give it the one yield it needs. */ + var later = await memo.GetAsync(T0.AddSeconds(10), scan.Run, CancellationToken.None); + Assert.Equal(1, scan.Calls); + Assert.Equal(10, later.AgeSeconds); + Assert.Equal(4, later.Counts[1].Healthy); + } + + /// The memo is per data source, not per process: the MCP host and the web host each build their + /// own , so each holds its own memo and "one scan per minute per HOST" is the + /// bound — and the same instance always maps to the same memo, or the single-flight would be no flight + /// at all. + [Fact] + public void TheMemo_IsOnePerDataSource() + { + using var mcpHost = NpgsqlDataSource.Create("Host=localhost;Username=mcp;Database=darling"); + using var webHost = NpgsqlDataSource.Create("Host=localhost;Username=viewer;Database=darling"); + + var mcpMemo = DarlingFleetReader.CollectionHealthMemoFor(mcpHost); + Assert.Same(mcpMemo, DarlingFleetReader.CollectionHealthMemoFor(mcpHost)); + Assert.NotSame(mcpMemo, DarlingFleetReader.CollectionHealthMemoFor(webHost)); + } + + /// The age travels: BuildRollup carries it onto the payload, defaults it to 0 for the pure + /// reduction tests that build cards by hand, and both JSON surfaces (the MCP tool and /api/fleet + /// serialize the same object through the same options) spell it collection_health_age_seconds, + /// trailing. + [Fact] + public void TheAge_RidesTheRollup_AndSerializesTrailing() + { + var fresh = DarlingFleetReader.BuildRollup(Array.Empty(), T0, T0.AddHours(-1), T0); + Assert.Equal(0, fresh.CollectionHealthAgeSeconds); + + var served = DarlingFleetReader.BuildRollup( + Array.Empty(), T0, T0.AddHours(-1), T0, collectionHealthAgeSeconds: 42); + Assert.Equal(42, served.CollectionHealthAgeSeconds); + + var json = JsonSerializer.Serialize(served, DarlingFleetReader.JsonOptions); + JsonAssert.Contains("\"collection_health_age_seconds\": 42", json); + /* Trailing: nothing an existing consumer indexes by position moved. */ + Assert.True( + json.IndexOf("\"collection_health_age_seconds\"", StringComparison.Ordinal) + > json.IndexOf("\"tags\"", StringComparison.Ordinal)); + Assert.EndsWith("\"collection_health_age_seconds\":42}", json, StringComparison.Ordinal); + } + + /// The reader reaches the 7-day statement ONLY through the memo — a second direct call site would + /// silently restore one copy of the fan-out per caller and no behavioural test against a fake scan could + /// see it. Source-anchored on the reader's own text. + [Fact] + public void TheReader_RunsTheCollectionHealthStatement_OnlyThroughTheMemo() + { + var reader = RepoFile.ReadRepoFileLf( + "Darling", "PerformanceMonitor.Darling.Service", "Mcp", "DarlingFleetReader.cs"); + + Assert.Contains( + "await CollectionHealthMemoFor(postgres).GetAsync(", + reader, + StringComparison.Ordinal); + + /* Every call of the statement's reader is the one inside the memo's scan lambda. */ + var callSites = Regex.Matches(reader, @"ReadFailingCollectorCountsAsync\(postgres,") + .Select(m => m.Index) + .ToList(); + Assert.Single(callSites); + var lineStart = reader.LastIndexOf('\n', callSites[0]) + 1; + var line = reader[lineStart..reader.IndexOf('\n', callSites[0])]; + Assert.Contains("(scanNow, scanToken) => ReadFailingCollectorCountsAsync(postgres, scanNow, scanToken)", line, StringComparison.Ordinal); + + /* And the statement text and its deadline are untouched by this change — the plan was never the problem. */ + Assert.Contains("FROM v_collection_log\nWHERE collection_time >= $1\nAND server_id <> 0\nGROUP BY server_id, collector_name", reader, StringComparison.Ordinal); + Assert.Contains( + "await using var command = postgres.CreateCommand(FleetCollectionHealthSql);\n command.CommandTimeout = McpCommandDeadlines.ReadSeconds;", + reader, + StringComparison.Ordinal); + } +} + +/// +/// Gated (DARLING_TEST_PG) end-to-end: three calls raced +/// against one real data source with three different windows — the exact shape of the #3735 photo — cost the +/// store ONE collection-health statement, and the age travels through the whole read. Plants nothing: the +/// assertions are about the memo the reader routed through and the age on the payload, neither of which +/// depends on what the store holds, so there is no sentinel row to scope to or clean up. +/// +[Collection("live-postgres")] +public sealed class FleetCollectionHealthMemoLivePostgresTests +{ + [Fact] + public async Task ThreeRacedOverviewCalls_CostOneCollectionHealthScan_AndTheAgeTravels() + { + var connectionString = Environment.GetEnvironmentVariable("DARLING_TEST_PG"); + Assert.SkipWhen(string.IsNullOrEmpty(connectionString), + "Set DARLING_TEST_PG to a Postgres connection string to run the live fleet-memo test."); + + var ct = TestContext.Current.CancellationToken; + using (var connection = new NpgsqlConnection(connectionString)) + { + await connection.OpenAsync(ct); + await PgMigrations.MigrateAsync(connection, ct); + } + + await using var postgres = NpgsqlDataSource.Create(connectionString!); + var memo = DarlingFleetReader.CollectionHealthMemoFor(postgres); + Assert.Equal(0, memo.ScansStarted); + + /* The photo: one caller, three windows, in parallel. */ + var now = DateTime.UtcNow; + var raced = await Task.WhenAll( + DarlingFleetReader.GetFleetOverviewAsync(postgres, now.AddHours(-1), now, now, cancellationToken: ct), + DarlingFleetReader.GetFleetOverviewAsync(postgres, now.AddHours(-6), now, now, cancellationToken: ct), + DarlingFleetReader.GetFleetOverviewAsync(postgres, now.AddHours(-24), now, now, cancellationToken: ct)); + + Assert.Equal(1, memo.ScansStarted); + Assert.All(raced, r => Assert.Equal(0, r.CollectionHealthAgeSeconds)); + + /* Half a minute on, still one statement, and the payload says how old that half of it is. */ + var later = now.AddSeconds(30); + var hit = await DarlingFleetReader.GetFleetOverviewAsync(postgres, later.AddHours(-1), later, later, cancellationToken: ct); + Assert.Equal(1, memo.ScansStarted); + Assert.Equal(30, hit.CollectionHealthAgeSeconds); + + /* Past the minute: the second statement of the test, age back to 0. */ + var expired = now.AddSeconds(61); + var reread = await DarlingFleetReader.GetFleetOverviewAsync(postgres, expired.AddHours(-1), expired, expired, cancellationToken: ct); + Assert.Equal(2, memo.ScansStarted); + Assert.Equal(0, reread.CollectionHealthAgeSeconds); + + /* A second data source over the SAME store is the web host beside the MCP host: its own memo, its own + first statement, nothing borrowed from the first. */ + await using var otherHost = NpgsqlDataSource.Create(connectionString!); + var other = await DarlingFleetReader.GetFleetOverviewAsync(otherHost, now.AddHours(-1), now, now, cancellationToken: ct); + Assert.Equal(0, other.CollectionHealthAgeSeconds); + Assert.Equal(1, DarlingFleetReader.CollectionHealthMemoFor(otherHost).ScansStarted); + Assert.Equal(2, memo.ScansStarted); + } +} diff --git a/Darling/PerformanceMonitor.Darling.Service/Mcp/DarlingFleetReader.cs b/Darling/PerformanceMonitor.Darling.Service/Mcp/DarlingFleetReader.cs index 3cbb51755b..12cf96d03e 100644 --- a/Darling/PerformanceMonitor.Darling.Service/Mcp/DarlingFleetReader.cs +++ b/Darling/PerformanceMonitor.Darling.Service/Mcp/DarlingFleetReader.cs @@ -10,6 +10,8 @@ using System.Collections.Generic; using System.Globalization; using System.Linq; +using System.Runtime.CompilerServices; +using System.Runtime.ExceptionServices; using System.Text; using System.Text.Json; using System.Text.Json.Serialization; @@ -439,6 +441,32 @@ AND server_id <> 0 /// The default depth of the worst-first "Needs attention" ranking. public const int DefaultWorstCount = 5; + /// How long a completed 7-day collection-health scan is served from memory before the next + /// caller re-reads it (#3735) — 60 seconds. The floor of what the rollup could report differently: it + /// counts per-collector RUNS over seven days, and the fastest collector cadence is one minute, so between + /// two reads a minute apart at most one run per collector has landed and no band the shared + /// ladder computes over a week of them can have moved. Shorter + /// would re-run a 7-day aggregate to learn nothing; longer would make the fleet's collector dots lag + /// behind get_collection_health, which reads the same rows uncached. See + /// for what the memo is protecting against. + internal static readonly TimeSpan CollectionHealthMemoLifetime = TimeSpan.FromSeconds(60); + + /// One per , weakly keyed so a + /// disposed data source takes its memo with it. Per data source rather than a bare static for two reasons: + /// the MCP host and the web host each build their OWN data source (DarlingMcpHostService / + /// DarlingWebHostService, different roles, different pools), so each host holds its own memo and + /// one scan per minute per HOST is the bound — not one per process, which would hand the web host a rollup + /// the mcp role read; and gated-live tests spin several stores in one process, and a static shared across + /// them would bleed one store's collection health into another's (the same reason + /// ComposeStoreAvailability keys its cache this way). + private static readonly ConditionalWeakTable s_collectionHealthMemos = new(); + + /// The memo the 7-day collection-health rollup for is served through + /// (#3735). Internal so a live test can ask the memo how many scans three racing overview calls actually + /// cost the store. + internal static CollectionHealthMemo CollectionHealthMemoFor(NpgsqlDataSource postgres) => + s_collectionHealthMemos.GetOrCreateValue(postgres); + /* ─────────────────────────── the read ─────────────────────────── */ /// @@ -469,7 +497,15 @@ public static async Task GetFleetOverviewAsync( var deadlocks = await ReadDeadlocksAsync(postgres, windowStartUtc, windowEndUtc, cancellationToken); var pgDeadlocks = await ReadPgDeadlocksAsync(postgres, windowStartUtc, windowEndUtc, cancellationToken); var lastCollection = await ReadLastCollectionAsync(postgres, now, cancellationToken); - var failingCollectors = await ReadFailingCollectorCountsAsync(postgres, now, cancellationToken); + /* #3735: the ONE read in this fan-out that does not depend on the caller's window — the 7-day + collection-health aggregate is the same statement whatever hours_back was — and therefore the one + that is single-flighted and memoized. Every other read above and below stays a per-call read: each + either carries the window or is a latest-snapshot DISTINCT ON that costs one index descent per + server. See CollectionHealthMemo for the production photo this answers. */ + var (failingCollectors, collectionHealthAgeSeconds) = await CollectionHealthMemoFor(postgres).GetAsync( + now, + (scanNow, scanToken) => ReadFailingCollectorCountsAsync(postgres, scanNow, scanToken), + cancellationToken); var tags = await ReadTagsAsync(postgres, cancellationToken); var tagForest = await ReadTagForestAsync(postgres, cancellationToken); /* #3368: the deadlock band's tiers, read ONCE per roll-up rather than per card. A store reload can @@ -513,7 +549,7 @@ ancient timestamp. */ windowEndUtc - windowStartUtc, deadlockTiers)); } - return BuildRollup(cards, now, windowStartUtc, windowEndUtc, worstCount, tagForest); + return BuildRollup(cards, now, windowStartUtc, windowEndUtc, worstCount, tagForest, collectionHealthAgeSeconds); } /// Builds one pre-banded card from a server's raw cross-server reads (pure — the reduction the WPF @@ -744,13 +780,18 @@ over. Null exactly when the PostgreSQL arm took no difference (#3539). */ /// Reduces the pre-banded cards to the fleet rollup — band counts, cross-server totals (summed from /// the cards), and the worst-first ranking. Pure so the reduction is unit-testable without a store. + /// How old the collection-health half of the cards is (#3735) — + /// 0 when the 7-day scan ran for this roll-up, else the memo's age; published as + /// collection_health_age_seconds. Trailing and defaulted so the pure-reduction tests that hand this + /// method cards they built themselves keep reading as fresh. public static FleetOverviewResult BuildRollup( IReadOnlyList cards, DateTime now, DateTime windowStartUtc, DateTime windowEndUtc, int worstCount = DefaultWorstCount, - IReadOnlyList? tags = null) + IReadOnlyList? tags = null, + int collectionHealthAgeSeconds = 0) { var healthy = 0; var warning = 0; @@ -849,6 +890,7 @@ attributes a real gap imprecisely. The PostgreSQL arm is covered AND tallied on AdditionalProblemCount = Math.Max(0, problems.Count - worst.Count), Cards = cards, Tags = tags ?? Array.Empty(), + CollectionHealthAgeSeconds = collectionHealthAgeSeconds, }; } @@ -1355,6 +1397,190 @@ internal readonly record struct PgCpuRow( /// against. Healthy + Failing is NOT it: STALE, WARNING, STOPPED, NO_PERMISSIONS and EXTENSION_MISSING rows /// are all banded collectors that are neither. internal readonly record struct CollectorCounts(int Healthy, int Failing, int Total, string? DeadlockBand = null, string? PgDeadlockBand = null); + + /// One completed 7-day collection-health scan (#3735): the per-server counts + /// produced and the nowUtc the scan was started with — + /// which is both the instant its $1 = now - 7 days window was cut from and the instant + /// collection_health_age_seconds is measured from. Read-only by type so a caller sharing the + /// dictionary with every other caller of the next minute cannot mutate it under them; the reader only ever + /// TryGetValues it. + internal sealed record CollectionHealthScan(IReadOnlyDictionary Counts, DateTime ReadAtUtc); + + /// + /// Single-flight plus a one-minute memo over the 7-day collection-health rollup (#3735). Concurrent callers + /// share ONE in-flight scan; a completed scan younger than is + /// served from memory. One instance per , through + /// . + /// + /// The production photo this answers. 2026-09-19 13:36Z, the largest production store: + /// was running THREE times concurrently — the identical statement, + /// the same start, one per get_fleet_overview call a single caller had raced in parallel with three + /// different hours_back values — each with two parallel workers, nine backends on + /// collection_log in pure I/O waits while the collectors' flush band was mid-COPY. The slowest + /// copy crossed the mcp role's server-side statement_timeout of 15 s; the caller's retry landed + /// after the flush drained and succeeded. Solo, the same plan runs in 680 ms as a healthy parallel + /// aggregate. The cap did its job — raising it would have hidden the fan-out — and the plan is not the + /// problem, so neither is touched here. What was wrong is that the rollup is WINDOW-INDEPENDENT: + /// hours_back bounds only the blocking and deadlock reads, and this read's $1 is always + /// now - 7 days, so three windows bought three copies of one answer. The web fleet page reaches the + /// same read through /api/fleet, so a browser's auto-refresh and an agent's tick stack the same way. + /// The fix is not to run the statement N times: one scan per minute per host, however many overview calls + /// race. + /// + /// The caller's nowUtc is the memo's clock. It is already the reader's freshness + /// reference (the instant generated_at carries) and both production callers pass + /// DateTime.UtcNow; using it rather than a private stopwatch means the age published beside the + /// payload is measured from the instant the payload itself publishes, and a test can drive expiry without + /// sleeping. A caller whose nowUtc is BEHIND the memo's is served the memo at age 0: the rollup is + /// not older than that caller's reference and there is nothing newer to read. A hit's $1 is up to a + /// minute older than a fresh scan's would have been, which at a 7-day window is nothing. + /// + /// The shared scan runs on , bounded by the 20 s command + /// deadline and the role's 15 s statement_timeout; a caller's own token releases only that + /// caller. Three reasons, in order of weight. The MCP surface never hands a token down at all — + /// documents that every tool calls its reader without one — so on the + /// surface the photo was taken the scan already ran to its deadline whatever the caller did. The web + /// surface DOES thread HttpContext.RequestAborted, and a tab that navigates away mid-refresh must not + /// discard a scan two other waiters, or the next tick, are about to use; the alternative — a linked token + /// that cancels when the last waiter leaves — turns an auto-refreshing browser into cancel-and-restart + /// churn, N sequential copies of the statement instead of N concurrent ones, the same pile-up in a + /// different shape. And the worst case under None is bounded to ONE orphaned statement per host per + /// minute, which is the bound this class promises anyway. So when the first caller cancels while two others + /// wait: that caller's throws to it at once, the statement + /// keeps running for the two, and its result is memoized for everyone who arrives in the next minute. + /// + /// A failed scan is not memoized, and cannot become an unobserved exception. The shared task + /// carries its outcome rather than faulting: waiters present when a scan fails rethrow it with its original + /// stack; the memo keeps whatever it last read successfully (still served while under a minute old), and + /// the next caller past that starts a fresh scan. A scan every waiter had abandoned before it failed + /// completes quietly — had the shared task faulted instead, that exception would have surfaced on the + /// finalizer thread as TaskScheduler.UnobservedTaskException, from a read nobody was waiting on. + /// + /// What this does NOT do. The other twelve reads in the overview are untouched: each carries the + /// caller's window or is a latest-snapshot read that costs one index descent per server. The SQL text, the + /// command deadline and the role's timeout are unchanged. And caller-side hygiene still applies: an agent + /// that wants three windows should call once with the widest and derive, or sequence the three — the + /// window-bound reads still cost what they cost, and the write band does not care about the caller's + /// reasons. + /// + internal sealed class CollectionHealthMemo + { + /// The scan's outcome as a VALUE — exactly one of the two is non-null — so the shared task + /// completes successfully whether the statement did or not (see the class summary's last paragraph). + /// The dispatch info, not the bare exception, so a waiter's rethrow keeps the statement's own stack + /// rather than acquiring the waiter's. + private sealed record ScanOutcome(CollectionHealthScan? Scan, ExceptionDispatchInfo? Failure); + + private readonly object _gate = new(); + + /// The scan the current callers share. Non-null and incomplete while a statement is running; + /// a COMPLETED task here is history (its success is in _latest, its failure is in + /// nothing), and the next caller who finds the memo stale starts a new one over it. + private Task? _inFlight; + + /// The newest scan that SUCCEEDED, or null before the first one has. Served while younger than + /// ; a failed scan never replaces it. + private CollectionHealthScan? _latest; + + private int _scansStarted; + + /// How many statements this memo has actually started, over its whole life — the figure a + /// live test compares against the number of overview calls it raced. Diagnostic; nothing reads it in + /// production. + internal int ScansStarted + { + get + { + lock (_gate) + { + return _scansStarted; + } + } + } + + /// + /// The 7-day collection-health counts as of — from memory when the memo is + /// under a minute old, from the scan already in flight when there is one, else from a scan this call + /// starts through — with how many whole seconds older than + /// the reading is (0 when this call's own scan produced it). + /// + /// The statement, as a function of the instant to cut the window from and the token + /// to run under. The memo, not the caller, decides the token (see the class summary), which is why the + /// seam takes one rather than closing over the caller's. Substitutable so a test can count executions + /// without a store. + /// Releases THIS caller's wait. It does not reach the statement. + internal async Task<(IReadOnlyDictionary Counts, int AgeSeconds)> GetAsync( + DateTime nowUtc, + Func>> scan, + CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(scan); + + Task shared; + TaskCompletionSource? lead = null; + lock (_gate) + { + if (_latest is { } latest && nowUtc - latest.ReadAtUtc < CollectionHealthMemoLifetime) + { + return (latest.Counts, AgeSeconds(nowUtc, latest)); + } + + if (_inFlight is not { IsCompleted: false }) + { + /* RunContinuationsAsynchronously: the waiters' continuations — the rest of thirteen + overview reads each — must not run inline on whichever thread completes the statement, + or the leader's connection callback would carry every joiner's card assembly. */ + lead = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + _inFlight = lead.Task; + _scansStarted++; + } + + shared = _inFlight; + } + + if (lead is not null) + { + /* Started OUTSIDE the lock so the statement's synchronous prefix (command construction) never + runs under it, and deliberately not awaited here: this caller is one waiter among any + number, and its own cancellation below must not be the scan's. RunScanAsync completes the + source in both arms and never throws, so the discarded task cannot fault. */ + _ = RunScanAsync(lead, nowUtc, scan); + } + + var outcome = await shared.WaitAsync(cancellationToken); + outcome.Failure?.Throw(); + var read = outcome.Scan!; + return (read.Counts, AgeSeconds(nowUtc, read)); + } + + private async Task RunScanAsync( + TaskCompletionSource lead, + DateTime nowUtc, + Func>> scan) + { + try + { + var counts = await scan(nowUtc, CancellationToken.None); + var read = new CollectionHealthScan(counts, nowUtc); + lock (_gate) + { + _latest = read; + } + + lead.SetResult(new ScanOutcome(read, null)); + } + catch (Exception ex) + { + lead.SetResult(new ScanOutcome(null, ExceptionDispatchInfo.Capture(ex))); + } + } + + /// Whole seconds from the scan's reference instant to , floored at + /// zero — a caller whose clock reads behind the scan's is not holding a reading from the future, it is + /// holding the freshest one there is. + private static int AgeSeconds(DateTime nowUtc, CollectionHealthScan read) => + (int)Math.Max(0, Math.Floor((nowUtc - read.ReadAtUtc).TotalSeconds)); + } } /// @@ -1737,4 +1963,17 @@ public sealed class FleetOverviewResult /// parent, ordering, and colour, so the fleet page can group cards under a nested tag tree even when a parent /// tag has no directly-assigned servers. Empty when no tags are defined. Assignment/editing stays desktop-only. [JsonPropertyName("tags")] public IReadOnlyList Tags { get; init; } = Array.Empty(); + + /// How many whole seconds older than the collection-health half of this + /// roll-up is (#3735): 0 when the 7-day collector-health scan ran for this call, otherwise the age of the + /// memoized scan it was served from — under 60 on a hit, because that is how long + /// DarlingFleetReader serves one scan before re-reading. The half it qualifies is every card's + /// healthy_collector_count / failed_collector_count / collector_count / + /// collector_severity / deadlock_collector_band / deadlock_source, and the fleet's + /// servers_with_collection_failures and deadlock_coverage; generated_at minus this is + /// that half's own reference instant. Everything else on the payload was read for this call. Published + /// so an agent that just watched a collector fail and re-reads the fleet a moment later knows why the dot + /// has not moved yet, and because a roll-up that serves a memo without saying so is a roll-up whose + /// freshness has to be trusted. Trailing and additive. + [JsonPropertyName("collection_health_age_seconds")] public int CollectionHealthAgeSeconds { get; init; } } diff --git a/Darling/PerformanceMonitor.Darling.Service/Mcp/DarlingMcpFleetTools.cs b/Darling/PerformanceMonitor.Darling.Service/Mcp/DarlingMcpFleetTools.cs index e4ab91ff90..bbb07bacbd 100644 --- a/Darling/PerformanceMonitor.Darling.Service/Mcp/DarlingMcpFleetTools.cs +++ b/Darling/PerformanceMonitor.Darling.Service/Mcp/DarlingMcpFleetTools.cs @@ -60,7 +60,15 @@ public sealed class DarlingMcpFleetTools "Unknown on purpose: PostgreSQL blocking is a once-a-minute SAMPLE of pg_stat_activity, and the " + "blocking band's count tiers were measured in engine-recorded reports per hour, so a sighting count " + "through them would band on a denominator they were never measured against — the PostgreSQL " + - "Blocking alert speaks for that condition until a sampled-shape band is measured.")] + "Blocking alert speaks for that condition until a sampled-shape band is measured. " + + "collection_health_age_seconds says how old the COLLECTOR-HEALTH half of the payload is: the per-card " + + "healthy/failed/collector_count, collector_severity, deadlock_collector_band and deadlock_source, and the " + + "fleet's servers_with_collection_failures and deadlock_coverage, come from one 7-day scan of the collection " + + "log that does not depend on hours_back and is shared by every overview call of the same minute on this " + + "host — 0 means this call ran it, anything up to 59 means a scan that many seconds before generated_at " + + "did. Everything else on the payload was read for this call. Racing several calls with different " + + "hours_back values buys nothing on that half and costs the window-bound reads N times over: call once " + + "with the widest window and derive, or sequence the calls.")] public static async Task GetFleetOverview( NpgsqlDataSource postgres, [Description("Hours of blocking/deadlock history the per-server cards and fleet totals window over. Default 1.")] int hours_back = 1) diff --git a/Darling/PerformanceMonitor.Darling.Service/Mcp/DarlingMcpInstructions.cs b/Darling/PerformanceMonitor.Darling.Service/Mcp/DarlingMcpInstructions.cs index e71b959215..ffe3b9201d 100644 --- a/Darling/PerformanceMonitor.Darling.Service/Mcp/DarlingMcpInstructions.cs +++ b/Darling/PerformanceMonitor.Darling.Service/Mcp/DarlingMcpInstructions.cs @@ -251,7 +251,7 @@ public static string Build(DarlingPeerDirectory.Snapshot peers) | Tool | Purpose | Key Parameters | |------|---------|----------------| | `get_sweep_reports` | The scheduled Fleet Sweep Reports — the stateful whole-fleet summaries persisted at the configured cadence (`get_alert_settings`' `fleet_sweep` group; default hourly): the sweep timeline over the window with each sweep's document embedded, the newest sweep in full (per-server verdicts with band transitions pre-computed, the would-have-paged ledger whenever the sweep ran under the alert master switch), and the watch-item worklist with its hysteresis position. Pass `sweep_id` for one sweep in full instead | `hours_back` (default 1), `as_of`, `sweep_id` (a STRING — the ids do not survive a JSON number round trip), `watch_state` (`pending`/`open`/`carried`/`closed`; omit for open + carried) | - | `get_fleet_overview` | Every enabled server's pre-banded health card (CPU, memory, blocking, deadlocks, worker threads, collector health) plus the fleet rollup — band counts, cross-server blocking / deadlock totals, and the worst-first "needs attention" ranking. Use it FIRST to decide which server to drill into | `hours_back` (default 1) | + | `get_fleet_overview` | Every enabled server's pre-banded health card (CPU, memory, blocking, deadlocks, worker threads, collector health) plus the fleet rollup — band counts, cross-server blocking / deadlock totals, and the worst-first "needs attention" ranking. Use it FIRST to decide which server to drill into. `collection_health_age_seconds` says how old the collector-health half is: that half is one 7-day scan shared by every overview call of the same minute, whatever `hours_back` was — so call once with the widest window rather than racing several | `hours_back` (default 1) | | `get_ag_health` | Always On Availability Group topology from the latest collection per server: each AG's replicas (role, connected / operational state, synchronization health, availability + failover mode, endpoint) and its per-database secondary state (synchronization state, log-send / redo queue KB, send / redo rate KB/s, estimated drain minutes, secondary lag seconds, suspended + reason) | `server_name` (optional — the whole fleet if omitted) | `get_ag_health` returns ONE GROUP PER (reporting server, AG). Every replica of an AG reports the whole AG's replica set, so an AG whose replicas are all monitored appears once per monitored replica, each group naming the `server_name` whose view it is. They are deliberately not merged: `operational_state` and `recovery_health` are populated only for the LOCAL replica, `connected_state` is meaningful only from the primary, and under WSFC quorum loss an instance answers from cached metadata — so the perspectives genuinely differ, and comparing them is how you spot a split view. A column the DMV does not populate reads null and bands `Unknown`, which never counts against a group's severity.