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 }; ///