Replace failure groups only for the batch's newest attempt

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<FailedMessageEntity> failedMessages, List<FailedMessageGroupEntity> 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<HashSet<Guid>> FindMessagesThisBatchIsNewestFor(List<FailedMessageEntity> 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

src/ServiceControl.Persistence.EFCore/Infrastructure/IIngestionSqlDialect.cs

@@ -18,8 +18,8 @@ public interface IIngestionSqlDialect
     Task UpsertFailedMessages(ServiceControlDbContext dbContext, IReadOnlyList<FailedMessageEntity> rows, CancellationToken cancellationToken);
 
     /// <summary>
-    /// 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.
     /// </summary>
     Task InsertGroups(ServiceControlDbContext dbContext, IReadOnlyList<FailedMessageGroupEntity> rows, CancellationToken cancellationToken);
 

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<FailedMessage.FailureGroup>
+            {
+                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()
     {

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()
     {

src/ServiceControl.Persistence.Tests/IngestedFailure.cs

@@ -105,7 +105,7 @@ Dictionary<string, string> BuildHeaders()
         }
     };
 
-    public IngestedFailure NextAttempt(DateTime attemptedAt) => new()
+    public IngestedFailure NextAttempt(DateTime attemptedAt, List<FailedMessage.FailureGroup> groups = null) => new()
     {
         MessageId = MessageId,
         EndpointName = EndpointName,
@@ -126,7 +126,7 @@ Dictionary<string, string> BuildHeaders()
         SendingEndpoint = SendingEndpoint,
         ReceivingEndpoint = ReceivingEndpoint,
         TimeSent = TimeSent,
-        Groups = Groups
+        Groups = groups ?? Groups
     };
 
     /// <summary>