From fe8e5d724e94cfd949d5b8189053ee1b4c694e84 Mon Sep 17 00:00:00 2001 From: John Simons Date: Tue, 4 Aug 2026 15:52:27 +1000 Subject: [PATCH] Replace failure groups only for the batch's newest attempt Ingestion replaces a message's failure group rows wholesale on every attempt, matching how the Raven persister assigns FailureGroups, so a message that fails again with a different exception moves to the group its latest attempt was classified into. The upsert only lets an attempt supply the payload columns when it is at least as new as the stored one, but the group replacement had no such guard. An older attempt arriving late from a concurrent writer therefore left the row describing one failure and the group rows describing another. Group replacement now applies only to the messages the batch is the newest attempt for, established by reading back the stored LastAttemptedAt inside the batch transaction, where the upsert already holds a row lock on every message involved. The read back compares with <= rather than ==, because PostgreSQL stores timestamps at microsecond precision while DateTime has 100ns ticks. Truncation is monotonic and applies to both sides of the upsert's own guard, so the two stay in step at any column precision. --- .../UnitOfWork/FailedMessageBatchWriter.cs | 49 ++++++++++++++++--- .../Infrastructure/IIngestionSqlDialect.cs | 4 +- .../EFCore/ErrorIngestionConcurrencyTests.cs | 42 ++++++++++++++++ .../EFCore/ErrorIngestionTests.cs | 22 +++++++++ .../IngestedFailure.cs | 4 +- 5 files changed, 110 insertions(+), 11 deletions(-) diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/FailedMessageBatchWriter.cs b/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/FailedMessageBatchWriter.cs index 247b19157c..646001b249 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/FailedMessageBatchWriter.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/FailedMessageBatchWriter.cs @@ -122,23 +122,58 @@ [.. knownEndpoints }) .DistinctBy(endpoint => endpoint.Id)]; - // Group rows are replaced wholesale on every attempt. When two concurrent writers process the - // same message their delete/insert pairs can interleave into a transient union of both - // attempts' groups; that is accepted, the next attempt replaces it. + // A message's groups are whatever its newest attempt classified it as, so they are replaced + // rather than merged. Only the messages this batch is now the newest attempt for take part: an + // older attempt arriving late from a concurrent writer already lost the payload columns in the + // upsert, and leaving it the groups would describe one failure in the row and another in the + // group rows. The upsert holds a row lock on every message in the batch until the transaction + // commits, so no competing writer can act on these messages between the delete and the insert. async Task ReplaceGroups(List failedMessages, List groups, CancellationToken cancellationToken) { - var messageIds = failedMessages.Select(message => message.UniqueMessageId).ToArray(); + var newestAttemptFor = await FindMessagesThisBatchIsNewestFor(failedMessages, cancellationToken); + + if (newestAttemptFor.Count == 0) + { + return; + } await dbContext.FailedMessageGroups - .Where(group => messageIds.Contains(group.FailedMessageUniqueId)) + .Where(group => newestAttemptFor.Contains(group.FailedMessageUniqueId)) .ExecuteDeleteAsync(cancellationToken); - if (groups.Count > 0) + var replacements = groups.Where(group => newestAttemptFor.Contains(group.FailedMessageUniqueId)).ToList(); + + if (replacements.Count > 0) { - await dialect.InsertGroups(dbContext, groups, cancellationToken); + await dialect.InsertGroups(dbContext, replacements, cancellationToken); } } + // The upsert has just stored the later of the incoming and the already stored attempt, so this + // batch is the newest attempt for exactly those messages whose stored value it now matches. + // + // Reading the value back rather than comparing against what the batch sent is what keeps this + // in step with the upsert's own guard on a provider whose column precision is coarser than + // DateTime: it truncates the stored value, which is why the comparison is <= and not ==. + async Task> FindMessagesThisBatchIsNewestFor(List failedMessages, CancellationToken cancellationToken) + { + var messageIds = failedMessages.Select(message => message.UniqueMessageId).ToArray(); + + var storedAttempts = await dbContext.FailedMessages + .AsNoTracking() + .Where(message => messageIds.Contains(message.UniqueMessageId)) + .Select(message => new { message.UniqueMessageId, message.LastAttemptedAt }) + .ToDictionaryAsync(row => row.UniqueMessageId, row => row.LastAttemptedAt, cancellationToken); + + return + [ + .. failedMessages + .Where(message => storedAttempts.TryGetValue(message.UniqueMessageId, out var storedAttempt) + && storedAttempt <= message.LastAttemptedAt) + .Select(message => message.UniqueMessageId) + ]; + } + async Task ResolveRetried(Guid[] retries, DateTime now, CancellationToken cancellationToken) { await dbContext.FailedMessages diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/IIngestionSqlDialect.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/IIngestionSqlDialect.cs index c94e75fa7f..7dd9fcd21a 100644 --- a/src/ServiceControl.Persistence.EFCore/Infrastructure/IIngestionSqlDialect.cs +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/IIngestionSqlDialect.cs @@ -18,8 +18,8 @@ public interface IIngestionSqlDialect Task UpsertFailedMessages(ServiceControlDbContext dbContext, IReadOnlyList rows, CancellationToken cancellationToken); /// - /// Insert if absent. The caller has already deleted the batch's messages' group rows in the - /// same transaction; if-absent keeps a concurrent writer's identical row from failing us. + /// Insert if absent. The caller has already deleted these messages' group rows in the same + /// transaction; if-absent keeps a concurrent writer's identical row from failing us. /// Task InsertGroups(ServiceControlDbContext dbContext, IReadOnlyList rows, CancellationToken cancellationToken); diff --git a/src/ServiceControl.Persistence.Tests/EFCore/ErrorIngestionConcurrencyTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/ErrorIngestionConcurrencyTests.cs index c265140b4d..4466341bbb 100644 --- a/src/ServiceControl.Persistence.Tests/EFCore/ErrorIngestionConcurrencyTests.cs +++ b/src/ServiceControl.Persistence.Tests/EFCore/ErrorIngestionConcurrencyTests.cs @@ -1,9 +1,11 @@ namespace ServiceControl.Persistence.Tests; using System; +using System.Collections.Generic; using System.Linq; using System.Threading.Tasks; using NUnit.Framework; +using ServiceControl.MessageFailures; using ServiceControl.Operations; class ErrorIngestionConcurrencyTests : ErrorIngestionTestBase @@ -42,6 +44,46 @@ await Task.WhenAll(Enumerable.Range(0, writers).Select(writer => Task.Run(async } } + [Test] + public async Task Concurrent_writers_leave_only_the_newest_attempts_groups() + { + const int writers = 4; + const int messages = 10; + + var seeds = Enumerable.Range(0, messages).Select(_ => new IngestedFailure()).ToArray(); + var baseTime = seeds[0].AttemptedAt; + + var groupsPerWriter = Enumerable.Range(0, writers) + .Select(writer => new List + { + new() { Id = Guid.NewGuid().ToString(), Title = $"Writer {writer}", Type = "Exception Type and Stack Trace" } + }) + .ToArray(); + + await Task.WhenAll(Enumerable.Range(0, writers).Select(writer => Task.Run(async () => + { + await using var unitOfWork = await UnitOfWorkFactory.StartNew(); + + foreach (var seed in seeds) + { + var attempt = seed.NextAttempt(baseTime.AddMinutes(writer), groupsPerWriter[writer]); + await unitOfWork.Recoverability.RecordFailedProcessingAttempt(attempt.Context, attempt.ProcessingAttempt, attempt.Groups); + } + + await unitOfWork.Complete(TestContext.CurrentContext.CancellationToken); + }))); + + var newestGroupId = groupsPerWriter[writers - 1][0].Id; + + foreach (var seed in seeds) + { + var groups = await GetGroups(seed.UniqueMessageId); + + Assert.That(groups.Select(group => group.GroupId), Is.EqualTo(new[] { newestGroupId }), + "only the newest attempt's groups may survive, regardless of commit order"); + } + } + [Test] public async Task Concurrent_writers_recording_the_same_endpoint_insert_it_once() { diff --git a/src/ServiceControl.Persistence.Tests/EFCore/ErrorIngestionTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/ErrorIngestionTests.cs index bb5c89598f..fa3f1160d2 100644 --- a/src/ServiceControl.Persistence.Tests/EFCore/ErrorIngestionTests.cs +++ b/src/ServiceControl.Persistence.Tests/EFCore/ErrorIngestionTests.cs @@ -213,6 +213,28 @@ public async Task An_older_attempt_counts_but_does_not_overwrite_the_newer_one() } } + [Test] + public async Task An_older_attempt_does_not_replace_the_newer_attempts_groups() + { + var newer = new IngestedFailure(); + await Ingest(newer); + + var older = new IngestedFailure + { + MessageId = newer.MessageId, + EndpointName = newer.EndpointName, + AttemptedAt = newer.AttemptedAt.AddMinutes(-5), + TimeOfFailure = newer.TimeOfFailure.AddMinutes(-5), + Groups = [new FailedMessage.FailureGroup { Id = Guid.NewGuid().ToString(), Title = "The older group", Type = "Exception Type and Stack Trace" }] + }; + await Ingest(older); + + var groups = await GetGroups(newer.UniqueMessageId); + + Assert.That(groups, Has.Count.EqualTo(1)); + Assert.That(groups[0].GroupId, Is.EqualTo(newer.Groups[0].Id), "The older attempt's groups replaced the newer attempt's"); + } + [Test] public async Task A_confirmed_retry_resolves_the_message_and_drops_its_retry_row() { diff --git a/src/ServiceControl.Persistence.Tests/IngestedFailure.cs b/src/ServiceControl.Persistence.Tests/IngestedFailure.cs index 472787f55a..a8395c5527 100644 --- a/src/ServiceControl.Persistence.Tests/IngestedFailure.cs +++ b/src/ServiceControl.Persistence.Tests/IngestedFailure.cs @@ -105,7 +105,7 @@ Dictionary BuildHeaders() } }; - public IngestedFailure NextAttempt(DateTime attemptedAt) => new() + public IngestedFailure NextAttempt(DateTime attemptedAt, List groups = null) => new() { MessageId = MessageId, EndpointName = EndpointName, @@ -126,7 +126,7 @@ Dictionary BuildHeaders() SendingEndpoint = SendingEndpoint, ReceivingEndpoint = ReceivingEndpoint, TimeSent = TimeSent, - Groups = Groups + Groups = groups ?? Groups }; ///