Order the failed message status by the times the events happened

src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlFailedMessageIngestionSqlDialect.cs

@@ -63,6 +63,36 @@ ON CONFLICT (id) DO NOTHING
         }
     }
 
+    public async Task ResolveRetriedMessages(ServiceControlDbContext dbContext, IReadOnlyList<ConfirmedRetry> rows, DateTime now, CancellationToken cancellationToken = default)
+    {
+        const int resolved = (int)FailedMessageStatus.Resolved;
+
+        foreach (var chunk in rows.Chunk(MaxRowsPerStatement))
+        {
+            await Execute(
+                dbContext,
+                $"""
+                 UPDATE failed_messages SET
+                     status = {resolved},
+                     status_changed_at = @p0,
+                     last_modified = @p0
+                 FROM (VALUES
+                 {ConfirmedRetryRows(chunk.Length)}
+                 ) AS s (unique_message_id, succeeded_at)
+                 WHERE failed_messages.unique_message_id = s.unique_message_id
+                   AND failed_messages.last_attempted_at <= s.succeeded_at
+                 """,
+                [[now], .. chunk.Select(retry => new object?[] { retry.UniqueMessageId, retry.SucceededAt })],
+                cancellationToken);
+        }
+    }
+
+    // A bare VALUES list has no target column to take its types from, so the first row carries them.
+    static string ConfirmedRetryRows(int rowCount) =>
+        string.Join(",\n", Enumerable.Range(0, rowCount).Select(row => row == 0
+            ? "(@p1::uuid, @p2::timestamptz)"
+            : $"(@p{1 + (row * 2)}, @p{2 + (row * 2)})"));
+
     // The columns the newer attempt wins wholesale
     static readonly string[] PayloadColumns =
     [
@@ -96,15 +126,20 @@ .. PayloadColumns
 
     static readonly string OnConflictUpdate = BuildOnConflictUpdate();
 
+    // Strictly newer, where the payload columns take the incoming attempt on a tie as well. A
+    // redelivery of the attempt already stored is not news, and must not undo a resolve or an
+    // archive that happened after it.
+    const string IsNewerAttempt = "excluded.last_attempted_at > failed_messages.last_attempted_at";
+
     static string BuildOnConflictUpdate()
     {
         const int unresolved = (int)FailedMessageStatus.Unresolved;
 
         var sql = new StringBuilder(
             $"""
              ON CONFLICT (unique_message_id) DO UPDATE SET
-                 status = {unresolved},
-                 status_changed_at = CASE WHEN failed_messages.status <> {unresolved} THEN excluded.status_changed_at ELSE failed_messages.status_changed_at END,
+                 status = CASE WHEN {IsNewerAttempt} THEN {unresolved} ELSE failed_messages.status END,
+                 status_changed_at = CASE WHEN {IsNewerAttempt} AND failed_messages.status <> {unresolved} THEN excluded.status_changed_at ELSE failed_messages.status_changed_at END,
                  last_modified = excluded.last_modified,
                  number_of_processing_attempts = failed_messages.number_of_processing_attempts
                      + CASE WHEN excluded.last_attempted_at <> failed_messages.last_attempted_at THEN excluded.number_of_processing_attempts ELSE 0 END,

src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerFailedMessageIngestionSqlDialect.cs

@@ -76,6 +76,35 @@ WHEN NOT MATCHED THEN INSERT ([Id], [Name], [HostId], [Host], [Monitored])
         }
     }
 
+    public async Task ResolveRetriedMessages(ServiceControlDbContext dbContext, IReadOnlyList<ConfirmedRetry> rows, DateTime now, CancellationToken cancellationToken = default)
+    {
+        const int resolved = (int)FailedMessageStatus.Resolved;
+
+        var maxRowsPerStatement = MaxRowsPerStatement(2);
+        foreach (var chunk in rows.Chunk(maxRowsPerStatement))
+        {
+            await Execute(
+                dbContext,
+                $"""
+                 UPDATE t SET
+                     [Status] = {resolved},
+                     [StatusChangedAt] = @p0,
+                     [LastModified] = @p0
+                 FROM [FailedMessages] AS t
+                 INNER JOIN (VALUES
+                 {ConfirmedRetryRows(chunk.Length)}
+                 ) AS s ([UniqueMessageId], [SucceededAt])
+                     ON t.[UniqueMessageId] = s.[UniqueMessageId]
+                 WHERE t.[LastAttemptedAt] <= s.[SucceededAt];
+                 """,
+                [[now], .. chunk.Select(retry => new object?[] { retry.UniqueMessageId, retry.SucceededAt })],
+                cancellationToken);
+        }
+    }
+
+    static string ConfirmedRetryRows(int rowCount) =>
+        string.Join(",\n", Enumerable.Range(0, rowCount).Select(row => $"(@p{1 + (row * 2)}, @p{2 + (row * 2)})"));
+
     // The columns the newer attempt wins wholesale
     static readonly string[] PayloadColumns =
     [
@@ -111,15 +140,20 @@ .. PayloadColumns
 
     static readonly string WhenMatchedUpdate = BuildWhenMatchedUpdate();
 
+    // Strictly newer, where the payload columns take the incoming attempt on a tie as well. A
+    // redelivery of the attempt already stored is not news, and must not undo a resolve or an
+    // archive that happened after it.
+    const string IsNewerAttempt = "s.[LastAttemptedAt] > t.[LastAttemptedAt]";
+
     static string BuildWhenMatchedUpdate()
     {
         const int unresolved = (int)FailedMessageStatus.Unresolved;
 
         var sql = new StringBuilder(
             $"""
              WHEN MATCHED THEN UPDATE SET
-                 [Status] = {unresolved},
-                 [StatusChangedAt] = CASE WHEN t.[Status] <> {unresolved} THEN s.[StatusChangedAt] ELSE t.[StatusChangedAt] END,
+                 [Status] = CASE WHEN {IsNewerAttempt} THEN {unresolved} ELSE t.[Status] END,
+                 [StatusChangedAt] = CASE WHEN {IsNewerAttempt} AND t.[Status] <> {unresolved} THEN s.[StatusChangedAt] ELSE t.[StatusChangedAt] END,
                  [LastModified] = s.[LastModified],
                  [NumberOfProcessingAttempts] = t.[NumberOfProcessingAttempts]
                      + CASE WHEN s.[LastAttemptedAt] <> t.[LastAttemptedAt] THEN s.[NumberOfProcessingAttempts] ELSE 0 END,

src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFIngestionUnitOfWork.cs

@@ -17,7 +17,7 @@ public class EFIngestionUnitOfWork : IIngestionUnitOfWork
     readonly ConcurrentQueue<RecordedFailedProcessingAttempt> failedProcessingAttempts = new();
     readonly ConcurrentQueue<Task> bodyWrites = new();
     readonly ConcurrentQueue<KnownEndpoint> knownEndpoints = new();
-    readonly ConcurrentQueue<Guid> confirmedRetries = new();
+    readonly ConcurrentQueue<ConfirmedRetry> confirmedRetries = new();
 
     public EFIngestionUnitOfWork(IAsyncDisposable scope, ServiceControlDbContext dbContext, IBodyStoragePersistence storagePersistence, EFPersisterSettings settings, IFailedMessageIngestionSqlDialect dialect, TimeProvider timeProvider)
     {
@@ -39,7 +39,7 @@ public EFIngestionUnitOfWork(IAsyncDisposable scope, ServiceControlDbContext dbC
 
     internal void Record(KnownEndpoint knownEndpoint) => knownEndpoints.Enqueue(knownEndpoint);
 
-    internal void RecordConfirmedRetry(Guid uniqueMessageId) => confirmedRetries.Enqueue(uniqueMessageId);
+    internal void RecordConfirmedRetry(ConfirmedRetry confirmedRetry) => confirmedRetries.Enqueue(confirmedRetry);
 
     public async Task Complete(CancellationToken cancellationToken = default)
     {

src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFRecoverabilityIngestionUnitOfWork.cs

@@ -61,9 +61,9 @@ public Task RecordFailedProcessingAttempt(MessageContext context,
         return Task.CompletedTask;
     }
 
-    public Task RecordSuccessfulRetry(string retriedMessageUniqueId, CancellationToken cancellationToken = default)
+    public Task RecordSuccessfulRetry(string retriedMessageUniqueId, DateTime succeededAt, CancellationToken cancellationToken = default)
     {
-        parentUnitOfWork.RecordConfirmedRetry(Guid.Parse(retriedMessageUniqueId));
+        parentUnitOfWork.RecordConfirmedRetry(new ConfirmedRetry(Guid.Parse(retriedMessageUniqueId), succeededAt));
 
         return Task.CompletedTask;
     }

src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/FailedMessageBatchWriter.cs

@@ -9,19 +9,19 @@ namespace ServiceControl.Persistence.EFCore.Implementation.UnitOfWork;
 // Writes one ingestion batch inside a single transaction. The statements providers genuinely
 // differ on (the upserts) come from the injected dialect; everything portable stays here as
 // set-based EF operations. Statement order matters: a message that fails and is retry-confirmed
-// in the same batch must end Resolved.
+// in the same batch must end Resolved, which the resolve running last is what gives it.
 class FailedMessageBatchWriter(ServiceControlDbContext dbContext, IFailedMessageIngestionSqlDialect dialect)
 {
     public async Task Write(
         IReadOnlyCollection<RecordedFailedProcessingAttempt> attempts,
         IReadOnlyCollection<KnownEndpoint> knownEndpoints,
-        IReadOnlyCollection<Guid> confirmedRetries,
+        IReadOnlyCollection<ConfirmedRetry> confirmedRetries,
         DateTime now,
         CancellationToken cancellationToken = default)
     {
         var (failedMessages, groups) = Fold(attempts, now);
         var endpoints = BuildEndpointRows(knownEndpoints);
-        var retries = confirmedRetries.Distinct().ToArray();
+        var retries = FoldRetries(confirmedRetries);
 
         if (failedMessages.Count == 0 && endpoints.Count == 0 && retries.Length == 0)
         {
@@ -110,6 +110,14 @@ await strategy.ExecuteAsync(async ct =>
         return (messages, groups);
     }
 
+    // A message acknowledged more than once in one batch keeps the latest acknowledgement, so the
+    // guard the resolve applies is the one that lets the most attempts through.
+    static ConfirmedRetry[] FoldRetries(IReadOnlyCollection<ConfirmedRetry> confirmedRetries) =>
+        [.. confirmedRetries
+            .GroupBy(retry => retry.UniqueMessageId)
+            .Select(group => new ConfirmedRetry(group.Key, group.Max(retry => retry.SucceededAt)))
+            .OrderBy(retry => retry.UniqueMessageId)];
+
     static List<KnownEndpointEntity> BuildEndpointRows(IReadOnlyCollection<KnownEndpoint> knownEndpoints) =>
         [.. knownEndpoints
             .Select(knownEndpoint => new KnownEndpointEntity
@@ -174,17 +182,17 @@ .. failedMessages
         ];
     }
 
-    async Task ResolveRetried(Guid[] retries, DateTime now, CancellationToken cancellationToken)
+    // The status only moves for messages whose newest stored attempt is not newer than the retry,
+    // which the dialect applies per row. The retry rows go regardless: the retry itself completed,
+    // so its claim is released whether or not the message has since failed again.
+    async Task ResolveRetried(ConfirmedRetry[] retries, DateTime now, CancellationToken cancellationToken)
     {
-        await dbContext.FailedMessages
-            .Where(failedMessage => retries.Contains(failedMessage.UniqueMessageId))
-            .ExecuteUpdateAsync(setters => setters
-                .SetProperty(failedMessage => failedMessage.Status, FailedMessageStatus.Resolved)
-                .SetProperty(failedMessage => failedMessage.StatusChangedAt, now)
-                .SetProperty(failedMessage => failedMessage.LastModified, now), cancellationToken);
+        await dialect.ResolveRetriedMessages(dbContext, retries, now, cancellationToken);
+
+        var messageIds = retries.Select(retry => retry.UniqueMessageId).ToArray();
 
         await dbContext.FailedMessageRetries
-            .Where(retry => retries.Contains(retry.UniqueMessageId))
+            .Where(retry => messageIds.Contains(retry.UniqueMessageId))
             .ExecuteDeleteAsync(cancellationToken);
     }
 }

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

@@ -0,0 +1,7 @@
+namespace ServiceControl.Persistence.EFCore.Infrastructure;
+
+/// <summary>
+/// A retry acknowledgement, carrying the time the retry succeeded so that a message which failed
+/// again afterwards is not resolved by it.
+/// </summary>
+public readonly record struct ConfirmedRetry(Guid UniqueMessageId, DateTime SucceededAt);

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

@@ -1,4 +1,4 @@
-namespace ServiceControl.Persistence.EFCore.Infrastructure;
+namespace ServiceControl.Persistence.EFCore.Infrastructure;
 
 using ServiceControl.Persistence.EFCore.DbContexts;
 using ServiceControl.Persistence.EFCore.Entities;
@@ -28,4 +28,11 @@ public interface IFailedMessageIngestionSqlDialect
     /// Insert if absent, never update: existing endpoints keep their Monitored flag.
     /// </summary>
     Task InsertMissingKnownEndpoints(ServiceControlDbContext dbContext, IReadOnlyList<KnownEndpointEntity> rows, CancellationToken cancellationToken = default);
+
+    /// <summary>
+    /// One row per message, distinct by UniqueMessageId. Marks each Resolved unless its stored
+    /// attempt is newer than the retry succeeded, which means the message failed again afterwards
+    /// and must stay Unresolved however the two batches interleave.
+    /// </summary>
+    Task ResolveRetriedMessages(ServiceControlDbContext dbContext, IReadOnlyList<ConfirmedRetry> rows, DateTime now, CancellationToken cancellationToken = default);
 }

src/ServiceControl.Persistence.RavenDB/ExpirationManager.cs

@@ -39,14 +39,21 @@ public void EnableExpiration(IAsyncDocumentSession session, EventLogItem eventLo
             session.Advanced.GetMetadataFor(eventLogItem)[Constants.Documents.Metadata.Expires] = expiresAt;
         }
 
-        public void EnableExpiration(PatchRequest request)
+        public void EnableExpiration(PatchRequest request) => request.Script += "\n" + EnableExpirationScript(request);
+
+        // Registers the value and hands back the statement, for scripts that only expire the
+        // document down one branch and so cannot have it appended to the end.
+        public string EnableExpirationScript(PatchRequest request)
         {
             var expiredAt = DateTime.UtcNow + errorRetentionPeriod;
 
-            request.Script += "\nthis['@metadata']['@expires'] = args.Expires;";
             request.Values.Add("Expires", expiredAt);
+
+            return "this['@metadata']['@expires'] = args.Expires;";
         }
 
-        public void CancelExpiration(PatchRequest request) => request.Script += "delete this['@metadata']['@expires']";
+        public void CancelExpiration(PatchRequest request) => request.Script += CancelExpirationScript;
+
+        public const string CancelExpirationScript = "delete this['@metadata']['@expires'];";
     }
 }

src/ServiceControl.Persistence.RavenDB/UnitOfWork/RavenRecoverabilityIngestionUnitOfWork.cs

@@ -68,20 +68,24 @@ public Task RecordFailedProcessingAttempt(
             return Task.CompletedTask;
         }
 
-        public Task RecordSuccessfulRetry(string retriedMessageUniqueId, CancellationToken cancellationToken = default)
+        public Task RecordSuccessfulRetry(string retriedMessageUniqueId, DateTime succeededAt, CancellationToken cancellationToken = default)
         {
             var failedMessageDocumentId = FailedMessageIdGenerator.MakeDocumentId(retriedMessageUniqueId);
             var failedMessageRetryDocumentId = RetryDocumentDataStore.MakeFailedMessageRetriesDocumentId(retriedMessageUniqueId);
 
             var patchRequest = new PatchRequest
             {
-                Script = $@"this.{nameof(FailedMessage.Status)} = {(int)FailedMessageStatus.Resolved};"
+                Values = new Dictionary<string, object> { { "succeededAt", succeededAt } }
             };
 
-            expirationManager.EnableExpiration(patchRequest);
+            patchRequest.Script = $@"if({NewestStoredAttempt("<=", "args.succeededAt")}){{
+                                        this.{nameof(FailedMessage.Status)} = {(int)FailedMessageStatus.Resolved};
+                                        {expirationManager.EnableExpirationScript(patchRequest)}
+                                     }}";
 
             parentUnitOfWork.AddCommand(new PatchCommandData(failedMessageDocumentId, null, patchRequest));
 
+            // The retry itself did complete, so its claim is released either way.
             parentUnitOfWork.AddCommand(new DeleteCommandData(failedMessageRetryDocumentId, null));
             return Task.CompletedTask;
         }
@@ -91,14 +95,24 @@ ICommandData CreateFailedMessagesPatchCommand(string uniqueMessageId, FailedMess
         {
             var documentId = FailedMessageIdGenerator.MakeDocumentId(uniqueMessageId);
 
-            const string ProcessingAttempts = nameof(FailedMessage.ProcessingAttempts);
-            const string AttemptedAt = nameof(FailedMessage.ProcessingAttempt.AttemptedAt);
-
             //HINT: RavenDB 4.2 removed Lodash utility functions, but supports ECMAScript 5.1 and some 6.0 features like arrow functions and array primitive functions
             var existingDocPatch = new PatchRequest
             {
-                Script = $@"this.{nameof(FailedMessage.Status)} = args.status;
-                                this.{nameof(FailedMessage.FailureGroups)} = args.failureGroups;
+                // The status, the groups and the retention stamp describe the message as its newest
+                // attempt left it, so an attempt that is not the newest only joins the attempts
+                // array. Which attempt is newest has to be read before the incoming one is pushed.
+                //
+                // A message re-failing must not keep an @expires stamp set by an earlier
+                // resolve/archive/retry, otherwise it can be silently deleted while still
+                // Unresolved. A late older attempt must not strip the stamp off a message that is
+                // still resolved, which is why that runs down the same branch.
+                Script = $@"var isNewestAttempt = {NewestStoredAttempt("<", $"args.attempt.{AttemptedAt}")};
+
+                                if(isNewestAttempt){{
+                                    this.{nameof(FailedMessage.Status)} = args.status;
+                                    this.{nameof(FailedMessage.FailureGroups)} = args.failureGroups;
+                                    {ExpirationManager.CancelExpirationScript}
+                                }}
 
                                 var newAttempts = this.{nameof(FailedMessage.ProcessingAttempts)};
 
@@ -127,10 +141,6 @@ ICommandData CreateFailedMessagesPatchCommand(string uniqueMessageId, FailedMess
                     },
             };
 
-            // A message re-failing must not keep an @expires stamp set by an earlier
-            // resolve/archive/retry, otherwise it can be silently deleted while still Unresolved.
-            expirationManager.CancelExpiration(existingDocPatch);
-
             return new PatchCommandData(documentId, null, existingDocPatch,
                 patchIfMissing: new PatchRequest
                 {
@@ -167,6 +177,14 @@ void AddStoreBodyCommands(MessageContext context, string contentType)
         static string GetContentType(IReadOnlyDictionary<string, string> headers, string defaultContentType)
             => headers.GetValueOrDefault(Headers.ContentType, defaultContentType);
 
+        // Attempts are stored sorted ascending by AttemptedAt, so the last one is the newest. Both
+        // sides are RavenDB serialized dates, compared as strings exactly as the sort above does.
+        static string NewestStoredAttempt(string comparison, string time) =>
+            $"this.{ProcessingAttempts}.length === 0 || this.{ProcessingAttempts}[this.{ProcessingAttempts}.length - 1].{AttemptedAt} {comparison} {time}";
+
+        const string ProcessingAttempts = nameof(FailedMessage.ProcessingAttempts);
+        const string AttemptedAt = nameof(FailedMessage.ProcessingAttempt.AttemptedAt);
+
         static int MaxProcessingAttempts = 10;
         // large object heap starts above 85000 bytes and not above 85 KB!
         internal const int LargeObjectHeapThreshold = 85_000;