diff --git a/Darling/Darling.Tests/RdsDeadlockIngestorTests.cs b/Darling/Darling.Tests/RdsDeadlockIngestorTests.cs
new file mode 100644
index 000000000..a60e5f0b9
--- /dev/null
+++ b/Darling/Darling.Tests/RdsDeadlockIngestorTests.cs
@@ -0,0 +1,300 @@
+/*
+ * 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";
+
+ /* The same report with a non-zero-offset zone in the prefix. PgDeadlockLogParser refuses the whole
+ read on this rather than storing local timestamps as UTC (#2993). */
+ private static readonly string NonUtcDeadlockText =
+ DeadlockText.Replace(" UTC ", " EST ", StringComparison.Ordinal);
+
+ /* 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);
+ }
+
+ ///
+ /// The parse-refusal case, and the one the defect was found through: a log stamped in a non-UTC zone
+ /// makes throw
+ /// (#2993), which abandons the whole read — siblings included. The marker must not move, because
+ /// log_timezone is a setting somebody can fix, and every report in that window is worth still
+ /// being there when they do.
+ ///
+ /// This is the shape the pre-#3008 order was worst for: the refusal is not a fault anybody
+ /// caused at collection time, it recurs every cycle until the GUC changes, and each occurrence consumed
+ /// a fresh window on the way past.
+ ///
+ [Fact]
+ public async Task ARefusedLogTimezoneDoesNotAdvanceTheMarker()
+ {
+ await using var store = NpgsqlDataSource.Create(DeadStore);
+ var (ingestor, client, logs) = Build(store, new FakeRds { FirstBody = NonUtcDeadlockText });
+
+ await Assert.ThrowsAsync(
+ () => ingestor.IngestAsync(1, "target-a", Host));
+
+ /* Still reachable, byte for byte — so the reports survive the setting being corrected. */
+ var again = await logs.ReadNewestAsync(Host);
+
+ Assert.Equal(NonUtcDeadlockText, again!.Value.Text);
+ Assert.Equal(2, client.Downloads.Count);
+ Assert.All(client.Downloads, d => Assert.Null(d.Marker));
+ }
+
+ ///
+ /// 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 f07961d77..57a489f8b 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/Darling.Tests/RdsPlanIngestorTests.cs b/Darling/Darling.Tests/RdsPlanIngestorTests.cs
new file mode 100644
index 000000000..c7c81bce6
--- /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);
+ }
+}
diff --git a/Darling/PerformanceMonitor.Darling.Service/Targets/RdsDeadlockIngestor.cs b/Darling/PerformanceMonitor.Darling.Service/Targets/RdsDeadlockIngestor.cs
index 0a9bd6c0e..0d1996d04 100644
--- a/Darling/PerformanceMonitor.Darling.Service/Targets/RdsDeadlockIngestor.cs
+++ b/Darling/PerformanceMonitor.Darling.Service/Targets/RdsDeadlockIngestor.cs
@@ -82,15 +82,52 @@ and empty. */
ex.Message, RdsLogUnavailableException.IsAuthorizationRefusal(ex), ex);
}
- if (chunk is null || string.IsNullOrEmpty(chunk.Value.Text))
+ if (chunk is null)
{
return 0;
}
- /* Not inside the tolerant catch above, which covers the AWS FETCH. A parse refusal is a statement
- about the target's configuration rather than about reaching it, and it has to reach the runner
- to be classified. */
- 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;
+ }
+
+ /* Not inside IngestAsync's tolerant catch, which covers the AWS FETCH. A parse refusal is a
+ statement about the target's configuration rather than about reaching it, and it has to reach the
+ runner to be classified.
+
+ It also has to leave WITHOUT the marker being committed, which is why this whole method sits
+ ahead of the commit rather than around it: a refused zone that consumed the window would discard
+ every report in it, and the setting that caused the refusal is fixable, so those reports are
+ worth still being there afterwards (#3008). */
+ 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 bcbe6de1e..459f8a1dc 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 3bb46b2c9..72e7cb1f4 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)
{