From f9f60cbdbe1893472054731eda34b752fe8b3c93 Mon Sep 17 00:00:00 2001 From: Erik Darling <2136037+erikdarlingdata@users.noreply.github.com> Date: Sat, 5 Sep 2026 08:46:34 -0400 Subject: [PATCH 1/2] Commit the RDS resume marker after the write, not inside the fetch (#3008) RdsLogSource recorded the marker as a side effect of DownloadDBLogFilePortion, before the caller had parsed or stored anything. The transport is consume-once and the marker is per-process, so any failure between the fetch and a committed COPY consumed a window nobody stored: a parse fault, a COPY past its deadline, a dropped store connection and a cancelled cycle each lost every report in that window with no error naming the loss. ReadNewestAsync now returns the candidate marker with the chunk, and both ingestors commit it once the rows are durable. The commit sits at one point per ingestor, after a StoreAsync that either stores everything the chunk held or throws, so the marker cannot advance over a partial result. Applied to the plan route as well as the deadlock route: both call ReadNewestAsync, each holds its own RdsLogSource, and the defect was identical in both. --- .../Darling.Tests/RdsDeadlockIngestorTests.cs | 267 ++++++++++++++++++ Darling/Darling.Tests/RdsLogSourceTests.cs | 80 +++++- .../Targets/RdsDeadlockIngestor.cs | 36 ++- .../Targets/RdsLogSource.cs | 65 ++++- .../Targets/RdsPlanIngestor.cs | 35 ++- 5 files changed, 466 insertions(+), 17 deletions(-) create mode 100644 Darling/Darling.Tests/RdsDeadlockIngestorTests.cs diff --git a/Darling/Darling.Tests/RdsDeadlockIngestorTests.cs b/Darling/Darling.Tests/RdsDeadlockIngestorTests.cs new file mode 100644 index 0000000000..1f204c4fe8 --- /dev/null +++ b/Darling/Darling.Tests/RdsDeadlockIngestorTests.cs @@ -0,0 +1,267 @@ +/* + * 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; +using System.Threading.Tasks; +using Amazon.RDS; +using Amazon.RDS.Model; +using Npgsql; +using PerformanceMonitor.Collectors; +using PerformanceMonitor.Darling.Service.Targets; +using Xunit; + +namespace Darling.Tests; + +/// +/// The RDS deadlock ingestor's resume-marker discipline (#3008): the marker advances only after the chunk's +/// rows are in the store, so a cycle that dies between the fetch and a committed COPY leaves the window to +/// be read again instead of consuming it. +/// +/// Why this needs its own suite rather than an assertion in +/// RdsLogSourceTests. can only be asked whether it advanced when +/// told to. Whether anything ever tells it to — and whether that happens before or after the write — is the +/// INGESTOR's decision, and it is the half that was wrong. A transport-level test passes just as well over +/// a caller that never commits at all, which would be a different permanent bug (every cycle re-reading one +/// window forever while looking like progress). +/// +/// What makes the failures real rather than simulated. The store is a data source pointed at a +/// closed loopback port, the form DarlingMcpAlertToolsTests already uses: the COPY genuinely fails, +/// at the connection, with no Postgres needed — so these run on the Windows build job alongside the +/// live-store suites rather than only where a real store exists. The log text is the verbatim PostgreSQL +/// 17.11 capture PgDeadlockLogParserTests is built on, parsed by the real +/// , so the rows reaching the write are the rows production would +/// build. +/// +public sealed class RdsDeadlockIngestorTests +{ + /* A store that cannot be written to. Port 1 on loopback refuses immediately, and Timeout=1 bounds the + attempt, so the write fails deterministically and fast rather than hanging a CI job. */ + private const string DeadStore = + "Host=127.0.0.1;Port=1;Username=none;Password=none;Database=none;Timeout=1"; + + private const string Host = "solo.abc123.us-east-1.rds.amazonaws.com"; + + /* Verbatim from a PostgreSQL 17.11 target, via PgDeadlockLogParserTests — including the %Q query id + running straight into ERROR: with no separator, which a hand-written fixture would not have. */ + private const string DeadlockText = + "2026-08-26 22:25:24.100 UTC [1549] 322048460535975151ERROR: deadlock detected\n" + + "2026-08-26 22:25:24.100 UTC [1549] 322048460535975151DETAIL: Process 1549 waits for ShareLock on transaction 809; blocked by process 1556.\n" + + "\tProcess 1556 waits for ShareLock on transaction 808; blocked by process 1549.\n" + + "\tProcess 1549: \n" + + "\tBEGIN; UPDATE dl SET v=v+1 WHERE id=1; SELECT pg_sleep(2); UPDATE dl SET v=v+1 WHERE id=2; COMMIT;\n" + + "\tProcess 1556: \n" + + "\tBEGIN; UPDATE dl SET v=v+1 WHERE id=2; SELECT pg_sleep(2); UPDATE dl SET v=v+1 WHERE id=1; COMMIT;\n" + + "2026-08-26 22:25:24.100 UTC [1549] 322048460535975151HINT: See server log for query details.\n"; + + /* A window with log traffic but no deadlock in it — the ordinary quiet cycle, and the positive control + for the commit firing at all. */ + private const string QuietText = + "2026-08-26 22:30:00.000 UTC [1600] LOG: checkpoint starting: time\n" + + "2026-08-26 22:30:12.000 UTC [1600] LOG: checkpoint complete: wrote 42 buffers\n"; + + /// + /// An RDS client that models the transport's ONE load-bearing property: what you get back depends on + /// the marker you send, and a marker is only handed out once. A fake that answered the same text + /// regardless of the marker could not tell a resumed read from a repeated one, which is the entire + /// distinction under test. + /// + private sealed class FakeRds : AmazonRDSClient + { + public FakeRds() : base(new Amazon.Runtime.BasicAWSCredentials("a", "b"), + Amazon.RegionEndpoint.USEast1) { } + + public List Downloads { get; } = new(); + + /// What the FIRST (unresumed) read returns. + public string FirstBody { get; init; } = DeadlockText; + + public override Task DescribeDBLogFilesAsync( + DescribeDBLogFilesRequest request, CancellationToken cancellationToken = default) + => Task.FromResult(new DescribeDBLogFilesResponse + { + DescribeDBLogFiles = new List + { + new() { LogFileName = "error/postgresql.log.2026-08-26-22", LastWritten = 9999 }, + }, + }); + + public override Task DownloadDBLogFilePortionAsync( + DownloadDBLogFilePortionRequest request, CancellationToken cancellationToken = default) + { + Downloads.Add(request); + + /* Consume-once: the unresumed read yields the window under test, and only a read that + presented MARKER-1 has moved past it. */ + var resumed = request.Marker == "MARKER-1"; + + return Task.FromResult(new DownloadDBLogFilePortionResponse + { + LogFileData = resumed ? QuietText : FirstBody, + Marker = resumed ? "MARKER-2" : "MARKER-1", + AdditionalDataPending = false, + }); + } + } + + private static (RdsDeadlockIngestor Ingestor, FakeRds Client, RdsLogSource Logs) Build( + NpgsqlDataSource store, FakeRds? client = null) + { + var fake = client ?? new FakeRds(); + var logs = new RdsLogSource(_ => fake); + return (new RdsDeadlockIngestor(store, logs), fake, logs); + } + + /// + /// The precondition every failure test below rests on, named so a broken fixture reports itself + /// instead of turning those tests into vacuous passes: this text really does parse to one deadlock, so + /// a cycle over it really does reach the COPY. A fixture that stopped parsing would make the ingestor + /// store nothing, which is a legitimate commit — the failure tests would go red for a reason that has + /// nothing to do with the marker. + /// + [Fact] + public void TheFixtureReallyParsesToOneDeadlock() + { + var found = PgDeadlockLogParser.Extract(DeadlockText); + + Assert.Single(found); + Assert.Equal(1549, found[0].VictimPid); + + /* And the control text really does not, so the quiet-cycle test below is testing a quiet cycle + rather than a second copy of the deadlock case. */ + Assert.Empty(PgDeadlockLogParser.Extract(QuietText)); + } + + /// + /// The assertion this fix exists for. The store write fails, so the marker must not move: the + /// next cycle's request carries no marker, which on this transport is the only way the same bytes can + /// be fetched twice. + /// + /// Moving the commit back inside the fetch — the pre-#3008 order — fails + /// Assert.Null(client.Downloads[1].Marker) below with MARKER-1, because the fetch would + /// have recorded the marker before the COPY ever ran. + /// + [Fact] + public async Task AFailedStoreWriteDoesNotAdvanceTheMarker() + { + await using var store = NpgsqlDataSource.Create(DeadStore); + var (ingestor, client, _) = Build(store); + + var failure = await Assert.ThrowsAnyAsync( + () => ingestor.IngestAsync(1, "target-a", Host)); + + /* The WRITE failed, not the read. RdsLogUnavailableException is the read's failure wrapper, and if + it arrived here the fetch never produced a chunk and nothing below would mean anything. */ + Assert.IsNotType(failure); + Assert.Single(client.Downloads); + + /* Second cycle. The marker never advanced, so this is a fresh unresumed read of the same window. */ + await Assert.ThrowsAnyAsync(() => ingestor.IngestAsync(1, "target-a", Host)); + + Assert.Equal(2, client.Downloads.Count); + Assert.Null(client.Downloads[1].Marker); + Assert.Equal(client.Downloads[0].Marker, client.Downloads[1].Marker); + } + + /// + /// The point of not advancing, stated as the consequence rather than the mechanism: the deadlock text + /// is offered to the parser AGAIN on the next cycle. This is what "nothing was lost" actually means, + /// and it is a claim about the bytes rather than about a dictionary key. + /// + [Fact] + public async Task TheSameLogTextIsOfferedAgainAfterAFailedWrite() + { + await using var store = NpgsqlDataSource.Create(DeadStore); + var (ingestor, client, logs) = Build(store); + + await Assert.ThrowsAnyAsync(() => ingestor.IngestAsync(1, "target-a", Host)); + await Assert.ThrowsAnyAsync(() => ingestor.IngestAsync(1, "target-a", Host)); + + /* Read the transport directly for the third pass: the ingestor cannot report what it fetched when + its write throws, and what matters is that the bytes are still reachable. */ + var third = await logs.ReadNewestAsync(Host); + + Assert.NotNull(third); + Assert.Equal(DeadlockText, third!.Value.Text); + Assert.Single(PgDeadlockLogParser.Extract(third.Value.Text)); + + /* Three fetches, none of them resumed — the window was never consumed. */ + Assert.Equal(3, client.Downloads.Count); + Assert.All(client.Downloads, d => Assert.Null(d.Marker)); + } + + /// + /// A cancelled cycle is the same category as a failed write, and the reason the fix is placed at the + /// marker rather than at any one exception type: shutdown mid-COPY loses the window just as + /// permanently as a parse fault, and nothing about it is an error anyone would investigate. + /// + [Fact] + public async Task ACancelledCycleDoesNotAdvanceTheMarker() + { + await using var store = NpgsqlDataSource.Create(DeadStore); + var (ingestor, client, _) = Build(store); + + using var cancelled = new CancellationTokenSource(); + await cancelled.CancelAsync(); + + await Assert.ThrowsAnyAsync( + () => ingestor.IngestAsync(1, "target-a", Host, cancelled.Token)); + + /* The read happened - the fake ignores the token - and the write did not. */ + Assert.Single(client.Downloads); + + await Assert.ThrowsAnyAsync(() => ingestor.IngestAsync(1, "target-a", Host)); + + Assert.Null(client.Downloads[1].Marker); + } + + /// + /// The control that stops all of the above passing over a no-op. A commit that never fires would + /// satisfy every "did not advance" assertion here while re-reading one window forever — so a cycle that + /// succeeds has to be shown advancing. + /// + /// A window with log traffic but no deadlock in it is the honest way to show it without a live + /// store: nothing is stored, nothing CAN be lost, so the marker is free to move — and it must, or a + /// quiet target would ask RDS for the same bounded tail every cycle forever. The dead store below is + /// never opened, which is itself part of the assertion. + /// + [Fact] + public async Task AQuietCycleAdvancesTheMarker() + { + await using var store = NpgsqlDataSource.Create(DeadStore); + var (ingestor, client, _) = Build(store, new FakeRds { FirstBody = QuietText }); + + var rows = await ingestor.IngestAsync(1, "target-a", Host); + Assert.Equal(0, rows); + + var more = await ingestor.IngestAsync(1, "target-a", Host); + Assert.Equal(0, more); + + /* Advanced: the second read resumed from the first read's marker instead of asking for the tail + again. */ + Assert.Equal(2, client.Downloads.Count); + Assert.Null(client.Downloads[0].Marker); + Assert.Equal("MARKER-1", client.Downloads[1].Marker); + Assert.Equal(0, client.Downloads[1].NumberOfLines); + } + + /// + /// A non-RDS host is skipped without touching the store or the marker — that target reads its log + /// through pg_read_file, and there is nothing here to report or to resume. + /// + [Fact] + public async Task ANonRdsHostIsSkipped() + { + await using var store = NpgsqlDataSource.Create(DeadStore); + var (ingestor, client, _) = Build(store); + + Assert.Equal(0, await ingestor.IngestAsync(1, "target-a", "db.internal.example.com")); + Assert.Empty(client.Downloads); + } +} diff --git a/Darling/Darling.Tests/RdsLogSourceTests.cs b/Darling/Darling.Tests/RdsLogSourceTests.cs index f07961d77d..57a489f8b8 100644 --- a/Darling/Darling.Tests/RdsLogSourceTests.cs +++ b/Darling/Darling.Tests/RdsLogSourceTests.cs @@ -176,16 +176,23 @@ public async Task TheNewestLogIsChosenByTimeNotByName() } /// - /// First read asks for a bounded TAIL; the next resumes from the marker. Without the tail, a first read - /// against a rotated multi-GB log would pull all of it across the network — #2565 measured 772 MB in - /// twenty seconds at capture-everything. + /// First read asks for a bounded TAIL; the next resumes from the marker, ONCE THE FIRST CHUNK HAS BEEN + /// COMMITTED. Without the tail, a first read against a rotated multi-GB log would pull all of it across + /// the network — #2565 measured 772 MB in twenty seconds at capture-everything. + /// + /// The CommitResume between the two reads is the part worth noticing: it used to be absent + /// because the read recorded its own marker, which is #3008. The pair below and + /// are each other's control — one shows + /// the marker still advances when the caller commits, the other that it does not when the caller + /// doesn't, and a fix that broke either would pass the other. /// [Fact] public async Task FirstReadIsBounded_ThenItResumesFromTheMarker() { var (source, client) = Build(); - await source.ReadNewestAsync("solo.abc123.us-east-1.rds.amazonaws.com"); + var first = await source.ReadNewestAsync("solo.abc123.us-east-1.rds.amazonaws.com"); + source.CommitResume(first!.Value.Resume); await source.ReadNewestAsync("solo.abc123.us-east-1.rds.amazonaws.com"); Assert.Null(client.Downloads[0].Marker); @@ -195,6 +202,71 @@ public async Task FirstReadIsBounded_ThenItResumesFromTheMarker() Assert.Equal(0, client.Downloads[1].NumberOfLines); } + /// + /// An UNCOMMITTED chunk does not move the marker: the second read asks RDS for the same window as the + /// first, tail request and all (#3008). + /// + /// This is the transport-level half of the data-loss fix. DownloadDBLogFilePortion is + /// consume-once and the marker lives in this process, so a marker recorded inside the fetch made every + /// failure after it — a parse fault, a COPY past its deadline, a dropped store connection, a cancelled + /// cycle — consume a window nobody stored, with no error naming the loss. Reading twice with no commit + /// in between is exactly the shape a failed cycle followed by the next cycle takes. + /// + [Fact] + public async Task TheMarkerDoesNotAdvanceUntilTheChunkIsCommitted() + { + var (source, client) = Build(); + + await source.ReadNewestAsync("solo.abc123.us-east-1.rds.amazonaws.com"); + await source.ReadNewestAsync("solo.abc123.us-east-1.rds.amazonaws.com"); + + Assert.Equal(2, client.Downloads.Count); + + /* The retry is byte-for-byte the first request: no marker, and the same bounded tail. Asserting + only Marker would pass over a read that had quietly switched to NumberOfLines = 0 and so asked + for a different window than the one that was lost. */ + Assert.Null(client.Downloads[1].Marker); + Assert.Equal(client.Downloads[0].Marker, client.Downloads[1].Marker); + Assert.Equal(client.Downloads[0].NumberOfLines, client.Downloads[1].NumberOfLines); + Assert.True(client.Downloads[1].NumberOfLines > 0); + } + + /// + /// The marker a chunk carries is the one RDS sent, and it is keyed to the file it came from — the + /// property that makes a log rotation start a fresh marker instead of resuming a new file at an old + /// file's offset. Handing the token back is a caller's only move, so it has to arrive intact. + /// + [Fact] + public async Task AChunkCarriesTheServiceMarkerKeyedToItsFile() + { + var (source, _) = Build(); + + var chunk = await source.ReadNewestAsync("solo.abc123.us-east-1.rds.amazonaws.com"); + + Assert.Equal("MARKER-1", chunk!.Value.Resume.Marker); + Assert.Contains("error/postgresql.log.2026-08-25-18", chunk.Value.Resume.Key, StringComparison.Ordinal); + Assert.Contains("solo", chunk.Value.Resume.Key, StringComparison.Ordinal); + } + + /// + /// Committing a chunk RDS sent no marker with is a no-op rather than a wedge or a crash. The API + /// answers this way on a read that reached the end of a file, and a caller commits unconditionally on + /// its success path — so this arrives in normal operation, not only from a fabricated token. + /// + [Fact] + public async Task CommittingAChunkWithNoServiceMarkerDoesNothing() + { + var (source, client) = Build(new FakeRds { NextMarker = null }); + + var chunk = await source.ReadNewestAsync("solo.abc123.us-east-1.rds.amazonaws.com"); + source.CommitResume(chunk!.Value.Resume); + source.CommitResume(default); + await source.ReadNewestAsync("solo.abc123.us-east-1.rds.amazonaws.com"); + + Assert.Null(client.Downloads[1].Marker); + Assert.True(client.Downloads[1].NumberOfLines > 0); + } + /// A non-RDS host is not this transport's problem: null, so the caller uses pg_read_file. [Fact] public async Task ANonRdsHostReturnsNull() diff --git a/Darling/PerformanceMonitor.Darling.Service/Targets/RdsDeadlockIngestor.cs b/Darling/PerformanceMonitor.Darling.Service/Targets/RdsDeadlockIngestor.cs index b8866180c6..ad849a351f 100644 --- a/Darling/PerformanceMonitor.Darling.Service/Targets/RdsDeadlockIngestor.cs +++ b/Darling/PerformanceMonitor.Darling.Service/Targets/RdsDeadlockIngestor.cs @@ -77,12 +77,44 @@ and empty. */ ex.Message, RdsLogUnavailableException.IsAuthorizationRefusal(ex), ex); } - if (chunk is null || string.IsNullOrEmpty(chunk.Value.Text)) + if (chunk is null) { return 0; } - var deadlocks = PgDeadlockLogParser.Extract(chunk.Value.Text); + var written = await StoreAsync(serverId, storageName, chunk.Value.Text, cancellationToken); + + /* THE MARKER MOVES HERE AND NOWHERE ELSE. Reaching this line means everything the chunk held is + either in the store or was nothing to store; anything else threw out of StoreAsync above and + left the marker where it was, so the next cycle asks RDS for the same window again. + + The order is the fix (#3008). While the marker advanced inside ReadNewestAsync, a parse fault, a + COPY that tripped its deadline, a dropped store connection or a cancelled cycle each consumed a + window nobody stored — and DownloadDBLogFilePortion does not hand the same bytes out twice, so + every deadlock in it was gone with no error naming the loss. */ + _logs.CommitResume(chunk.Value.Resume); + + return written; + } + + /// + /// Parse a chunk and store what it held, or throw. Split out so the resume marker has exactly one + /// commit point above it: every way this can decline to store rows — empty text, a slab with no + /// deadlocks in it — is a legitimate zero that loses nothing, and every way it can FAIL leaves via an + /// exception rather than a zero the caller would have to tell apart from those. + /// + private async Task StoreAsync( + int serverId, + string storageName, + string text, + CancellationToken cancellationToken) + { + if (string.IsNullOrEmpty(text)) + { + return 0; + } + + var deadlocks = PgDeadlockLogParser.Extract(text); if (deadlocks.Count == 0) { diff --git a/Darling/PerformanceMonitor.Darling.Service/Targets/RdsLogSource.cs b/Darling/PerformanceMonitor.Darling.Service/Targets/RdsLogSource.cs index bcbe6de1e0..459f8a1dc3 100644 --- a/Darling/PerformanceMonitor.Darling.Service/Targets/RdsLogSource.cs +++ b/Darling/PerformanceMonitor.Darling.Service/Targets/RdsLogSource.cs @@ -38,6 +38,14 @@ namespace PerformanceMonitor.Darling.Service.Targets; /// overlapping window produces the same shapes rather than duplicates — the same property the /// pg_read_file route already relies on. Persisting the marker would buy nothing and add a schema /// rung that could disagree with reality after a log rotation. +/// +/// The marker only moves when the caller says so (), because that +/// re-read tolerance is the whole reason it is safe to prefer a repeat over a loss. The transport is +/// consume-once — DownloadDBLogFilePortion will not hand the same bytes out twice — so a marker that +/// advanced inside the fetch turned any later failure into permanent data loss, while a marker that +/// advances after the write can at worst re-store a window the store already tolerates. That is the same +/// trade the in-memory choice above already makes across a restart; this makes the in-process behaviour +/// match it instead of being strictly worse than it. /// public sealed class RdsLogSource { @@ -59,7 +67,46 @@ public RdsLogSource(Func? clientFactory = null) /// Raw log text, to be handed to PgPlanLogParser.Extract unchanged. /// RDS had more than one call's worth. The caller decides whether to keep /// pulling; this type does not loop, so one cycle cannot spend unbounded time on one target. - public readonly record struct LogChunk(string Text, bool MoreAvailable); + /// Where the NEXT read should start, once has actually reached + /// the store. Handed back rather than recorded on the way out — see + /// . + public readonly record struct LogChunk(string Text, bool MoreAvailable, ResumeMarker Resume); + + /// + /// A position this source can resume from, and the file it belongs to. Opaque to the caller: the + /// marker is a service token, not an offset, so there is nothing to compute with — a caller's only + /// move is to hand it back once the chunk it came with is durable. + /// + /// The (instance, file) this marker belongs to. + /// RDS's own resume token, or null when the response carried none. + public readonly record struct ResumeMarker(string? Key, string? Marker); + + /// + /// Advance this source past a chunk whose rows are in the store. + /// + /// This is separate from the read on purpose, and it is the whole point of the type. The + /// marker used to be recorded inside , before the caller had done anything + /// with the text. On a consume-once transport that made every failure between the fetch and a committed + /// COPY a permanent loss: DownloadDBLogFilePortion does not hand the same bytes out twice, the + /// marker lives in this process, and the next call resumed past a chunk nobody stored. A parse fault, a + /// COPY that tripped its deadline, a dropped store connection and a cancelled cycle all lost every + /// report in that window with no error naming the loss. + /// + /// Doing nothing on an unset marker is deliberate rather than defensive: it lets a caller commit + /// unconditionally on its success path without first asking whether the read produced a token, which is + /// the shape that keeps the commit next to the write it depends on. + /// + public void CommitResume(ResumeMarker resume) + { + if (string.IsNullOrEmpty(resume.Key) || string.IsNullOrEmpty(resume.Marker)) + { + return; + } + + /* Keyed by FILE as well as instance, so a log rotation starts a fresh marker instead of resuming a + new file at an old file's offset. */ + _markers[resume.Key] = resume.Marker; + } /// /// The newest PostgreSQL log file's unread portion, or null when this target is not RDS at all. @@ -114,16 +161,16 @@ public RdsLogSource(Func? clientFactory = null) }, cancellationToken); - if (!string.IsNullOrEmpty(response.Marker)) - { - /* Keyed by FILE as well as instance, so a log rotation starts a fresh marker instead of - resuming a new file at an old file's offset. */ - _markers[key] = response.Marker; - } + /* The marker is RETURNED, not recorded. Recording it here would advance this source past text the + caller has not looked at yet, and on a consume-once transport that is a permanent loss rather + than a repeated read — see CommitResume. */ /* AdditionalDataPending is bool? in the SDK. Treated as false when null: claiming more is - pending when the API did not say so would make a caller loop for data that is not there. */ - return new LogChunk(response.LogFileData ?? string.Empty, response.AdditionalDataPending == true); + pending when the API did not say so would make a caller loop for data that is not there. */ + return new LogChunk( + response.LogFileData ?? string.Empty, + response.AdditionalDataPending == true, + new ResumeMarker(key, response.Marker)); } /* An AWS SDK response collection is NULL when the service omitted it, not an empty list, so the two diff --git a/Darling/PerformanceMonitor.Darling.Service/Targets/RdsPlanIngestor.cs b/Darling/PerformanceMonitor.Darling.Service/Targets/RdsPlanIngestor.cs index 3bb46b2c9c..72e7cb1f41 100644 --- a/Darling/PerformanceMonitor.Darling.Service/Targets/RdsPlanIngestor.cs +++ b/Darling/PerformanceMonitor.Darling.Service/Targets/RdsPlanIngestor.cs @@ -86,12 +86,43 @@ What changes is that the cycle now says WHICH kind of nothing it found. */ ex.Message, RdsLogUnavailableException.IsAuthorizationRefusal(ex), ex); } - if (chunk is null || string.IsNullOrEmpty(chunk.Value.Text)) + if (chunk is null) { return 0; } - var plans = PgPlanLogParser.Extract(chunk.Value.Text); + var written = await StoreAsync(serverId, storageName, chunk.Value.Text, cancellationToken); + + /* THE MARKER MOVES HERE AND NOWHERE ELSE — the same order, and for the same reason, as + RdsDeadlockIngestor (#3008). Reaching this line means everything the chunk held is either in the + store or was nothing to store; anything else threw out of StoreAsync and left the marker where it + was, so the next cycle asks RDS for the same window again rather than resuming past it. + + Plan rows dedup on (queryid, plan_hash), so the repeat this can cause costs a re-store of shapes + the store already has. The loss it replaces was unbounded and silent. */ + _logs.CommitResume(chunk.Value.Resume); + + return written; + } + + /// + /// Parse a chunk and store what it held, or throw. Split out so the resume marker has exactly one + /// commit point above it: every way this can decline to store rows — empty text, a slab no plan + /// threshold was crossed in — is a legitimate zero that loses nothing, and every way it can FAIL leaves + /// via an exception rather than a zero the caller would have to tell apart from those. + /// + private async Task StoreAsync( + int serverId, + string storageName, + string text, + CancellationToken cancellationToken) + { + if (string.IsNullOrEmpty(text)) + { + return 0; + } + + var plans = PgPlanLogParser.Extract(text); if (plans.Count == 0) { From 954748e2013fbd53980510213432207c0eb1ff0d Mon Sep 17 00:00:00 2001 From: Erik Darling <2136037+erikdarlingdata@users.noreply.github.com> Date: Sat, 5 Sep 2026 09:04:10 -0400 Subject: [PATCH 2/2] Cover the plan ingestor's marker ordering too RdsPlanIngestor got the identical #3008 fix but had no test reaching IngestAsync: RdsPlanIngestionFromRealLogTests only exercises PgPlanLogParser. The two ingestors hold separate RdsLogSource instances and separate copies of the read-store-commit sequence, so the ordering can regress in one while the other's suite stays green - and folding the two StoreAsync copies into a shared helper is the edit most likely to do it. Mirrors the deadlock suite's shape rather than transcribing it: failed store write, cancellation, and the quiet-cycle control that a never-firing commit would fail. Uses the real auto_explain capture the plan parser's own suite reads. --- Darling/Darling.Tests/RdsPlanIngestorTests.cs | 184 ++++++++++++++++++ 1 file changed, 184 insertions(+) create mode 100644 Darling/Darling.Tests/RdsPlanIngestorTests.cs diff --git a/Darling/Darling.Tests/RdsPlanIngestorTests.cs b/Darling/Darling.Tests/RdsPlanIngestorTests.cs new file mode 100644 index 0000000000..c7c81bce6d --- /dev/null +++ b/Darling/Darling.Tests/RdsPlanIngestorTests.cs @@ -0,0 +1,184 @@ +/* + * 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.IO; +using System.Threading; +using System.Threading.Tasks; +using Amazon.RDS; +using Amazon.RDS.Model; +using Npgsql; +using PerformanceMonitor.Collectors; +using PerformanceMonitor.Darling.Service.Targets; +using Xunit; + +namespace Darling.Tests; + +/// +/// The RDS PLAN ingestor's resume-marker discipline (#3008) — the same invariant +/// pins for the deadlock route, on the other caller of +/// RdsLogSource.ReadNewestAsync. +/// +/// Why this is a second suite and not a line in the first. The two ingestors hold separate +/// instances and separate copies of the read-store-commit sequence, so the +/// ordering can regress in one without touching the other. The obvious future edit — folding the two +/// StoreAsync copies into a shared helper — is exactly the one that could put the commit back on the +/// wrong side of the write here while the deadlock suite stayed green. RdsPlanIngestionFromRealLogTests +/// covers PgPlanLogParser and never reaches IngestAsync, so before this file the plan half of +/// the fix had no regression test at all. +/// +/// Deliberately a subset rather than a transcription: the parse-refusal case has no plan-route +/// equivalent (the non-UTC refusal is the deadlock parser's), and the reasoning behind the fake and the +/// closed-port store is recorded once, on the deadlock suite. +/// +public sealed class RdsPlanIngestorTests +{ + private const string DeadStore = + "Host=127.0.0.1;Port=1;Username=none;Password=none;Database=none;Timeout=1"; + + private const string Host = "solo.abc123.us-east-1.rds.amazonaws.com"; + + /* The REAL auto_explain capture the plan parser's own suite is built on, read from the same place it + reads it. A hand-written plan block would agree with whatever I believed auto_explain's shape was. */ + private static readonly string PlanText = + File.ReadAllText(Path.Combine(AppContext.BaseDirectory, "Fixtures", "auto_explain_real_block.txt")); + + /* Log traffic with no plan in it — a server whose auto_explain threshold nothing crossed. */ + private const string QuietText = + "2026-08-26 22:30:00.000 UTC [1600] LOG: checkpoint starting: time\n" + + "2026-08-26 22:30:12.000 UTC [1600] LOG: checkpoint complete: wrote 42 buffers\n"; + + private sealed class FakeRds : AmazonRDSClient + { + public FakeRds() : base(new Amazon.Runtime.BasicAWSCredentials("a", "b"), + Amazon.RegionEndpoint.USEast1) { } + + public List Downloads { get; } = new(); + + public string FirstBody { get; init; } = PlanText; + + public override Task DescribeDBLogFilesAsync( + DescribeDBLogFilesRequest request, CancellationToken cancellationToken = default) + => Task.FromResult(new DescribeDBLogFilesResponse + { + DescribeDBLogFiles = new List + { + new() { LogFileName = "error/postgresql.log.2026-08-26-15", LastWritten = 9999 }, + }, + }); + + public override Task DownloadDBLogFilePortionAsync( + DownloadDBLogFilePortionRequest request, CancellationToken cancellationToken = default) + { + Downloads.Add(request); + + var resumed = request.Marker == "MARKER-1"; + + return Task.FromResult(new DownloadDBLogFilePortionResponse + { + LogFileData = resumed ? QuietText : FirstBody, + Marker = resumed ? "MARKER-2" : "MARKER-1", + AdditionalDataPending = false, + }); + } + } + + private static (RdsPlanIngestor Ingestor, FakeRds Client, RdsLogSource Logs) Build( + NpgsqlDataSource store, FakeRds? client = null) + { + var fake = client ?? new FakeRds(); + var logs = new RdsLogSource(_ => fake); + return (new RdsPlanIngestor(store, logs), fake, logs); + } + + /// + /// The precondition the failure tests rest on: the fixture really does parse to a plan, so a cycle over + /// it really does reach the COPY. A fixture that stopped parsing would make the ingestor store nothing — + /// a legitimate commit — and the tests below would fail for an unrelated reason. + /// + [Fact] + public void TheFixtureReallyParsesToAPlan() + { + Assert.Single(PgPlanLogParser.Extract(PlanText)); + Assert.Empty(PgPlanLogParser.Extract(QuietText)); + } + + /// + /// The plan route's copy of the assertion this fix exists for: the store write fails, so the marker + /// must not move, and the next cycle asks RDS for the same window rather than resuming past it. + /// + /// Moving the commit back inside the fetch fails Assert.Null(client.Downloads[1].Marker) + /// with MARKER-1. + /// + [Fact] + public async Task AFailedStoreWriteDoesNotAdvanceTheMarker() + { + await using var store = NpgsqlDataSource.Create(DeadStore); + var (ingestor, client, logs) = Build(store); + + var failure = await Assert.ThrowsAnyAsync( + () => ingestor.IngestAsync(1, "target-a", Host)); + + Assert.IsNotType(failure); + Assert.Single(client.Downloads); + + await Assert.ThrowsAnyAsync(() => ingestor.IngestAsync(1, "target-a", Host)); + + Assert.Equal(2, client.Downloads.Count); + Assert.Null(client.Downloads[1].Marker); + + /* And the bytes are genuinely still there, which is what "nothing was lost" means. */ + var third = await logs.ReadNewestAsync(Host); + Assert.Equal(PlanText, third!.Value.Text); + Assert.Single(PgPlanLogParser.Extract(third.Value.Text)); + } + + /// + /// Cancellation is the same category as a failed write — shutdown mid-COPY consumed the window just as + /// permanently, and nothing about it looked like an error worth investigating. + /// + [Fact] + public async Task ACancelledCycleDoesNotAdvanceTheMarker() + { + await using var store = NpgsqlDataSource.Create(DeadStore); + var (ingestor, client, _) = Build(store); + + using var cancelled = new CancellationTokenSource(); + await cancelled.CancelAsync(); + + await Assert.ThrowsAnyAsync( + () => ingestor.IngestAsync(1, "target-a", Host, cancelled.Token)); + + Assert.Single(client.Downloads); + + await Assert.ThrowsAnyAsync(() => ingestor.IngestAsync(1, "target-a", Host)); + + Assert.Null(client.Downloads[1].Marker); + } + + /// + /// The control that stops the two tests above passing over a commit that never fires: a cycle with + /// nothing to store loses nothing, so the marker must advance — otherwise a server below its + /// auto_explain threshold would re-request the same bounded tail every cycle forever. + /// + [Fact] + public async Task AQuietCycleAdvancesTheMarker() + { + await using var store = NpgsqlDataSource.Create(DeadStore); + var (ingestor, client, _) = Build(store, new FakeRds { FirstBody = QuietText }); + + Assert.Equal(0, await ingestor.IngestAsync(1, "target-a", Host)); + Assert.Equal(0, await ingestor.IngestAsync(1, "target-a", Host)); + + Assert.Equal(2, client.Downloads.Count); + Assert.Null(client.Downloads[0].Marker); + Assert.Equal("MARKER-1", client.Downloads[1].Marker); + Assert.Equal(0, client.Downloads[1].NumberOfLines); + } +}