From 3f7d44b9fed3b1e77c87f4d34e2372fb0c404149 Mon Sep 17 00:00:00 2001 From: John Simons Date: Fri, 21 Aug 2026 11:33:22 +1000 Subject: [PATCH] Order the failed message status by the times the events happened A retry acknowledgement set Status to Resolved unconditionally, and an incoming processing attempt set it to Unresolved unconditionally. Inside one batch the statement order settles it, and FailedMessageBatchWriter says so. Across two transactions the last commit wins, whichever event actually happened first. --error-ingestion-only made that reachable: several hosts drain the same error queue into one database, so a confirmation and a later failure of the same message can be on different hosts entirely. A message that failed again after its retry succeeded could end up sitting at Resolved, and a redelivered attempt could pull an already resolved or archived message back to Unresolved. The retry acknowledgement carries the time the retry succeeded, so RecordSuccessfulRetry now takes it. Both persisters compare it against the newest attempt they have stored: - a confirmation resolves only a message whose newest stored attempt is no newer than the retry - an attempt moves the status only when it is strictly newer than the newest stored attempt, where the payload columns already take a tie Strict on one side and not the other is deliberate. Refreshing the payload from a redelivery of the newest attempt changes nothing; letting that same redelivery flip the status undoes a resolve or an archive that came after it. For EF the resolve becomes a dialect statement, because the guard is per row and ExecuteUpdate has nowhere to put it. The retry rows are still deleted whatever the guard decides: the retry itself completed, so its claim is released either way. For RavenDB the two patch scripts grow a branch, which the retention stamp moves inside: a late older attempt must not strip @expires off a message that is still resolved, and ExpirationManager now hands out the statement so it can sit in a branch rather than only at the end of a script. MessageExpiryTests gave both of its attempts the same AttemptedAt, which models a redelivery rather than a re-failure. They are now a minute apart, which is what the test name describes. --- ...tgreSqlFailedMessageIngestionSqlDialect.cs | 39 ++++++++- ...lServerFailedMessageIngestionSqlDialect.cs | 38 ++++++++- .../UnitOfWork/EFIngestionUnitOfWork.cs | 4 +- .../EFRecoverabilityIngestionUnitOfWork.cs | 4 +- .../UnitOfWork/FailedMessageBatchWriter.cs | 30 ++++--- .../Infrastructure/ConfirmedRetry.cs | 7 ++ .../IFailedMessageIngestionSqlDialect.cs | 9 ++- .../ExpirationManager.cs | 13 ++- .../RavenRecoverabilityIngestionUnitOfWork.cs | 42 +++++++--- .../Expiration/MessageExpiryTests.cs | 13 ++- .../RetryConfirmationOrderingTests.cs | 79 +++++++++++++++++++ ...itFailedMessagesDataStoreRetentionTests.cs | 2 - .../EFCore/ErrorIngestionConcurrencyTests.cs | 70 ++++++++++++++++ .../EFCore/ErrorIngestionTestBase.cs | 8 +- .../EFCore/ErrorIngestionTests.cs | 2 +- .../EFCore/RetentionSweepTests.cs | 2 - .../IRecoverabilityIngestionUnitOfWork.cs | 9 ++- .../Operations/RetryConfirmationProcessor.cs | 25 +++++- 18 files changed, 348 insertions(+), 48 deletions(-) create mode 100644 src/ServiceControl.Persistence.EFCore/Infrastructure/ConfirmedRetry.cs create mode 100644 src/ServiceControl.Persistence.Tests.RavenDB/Recoverability/RetryConfirmationOrderingTests.cs diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlFailedMessageIngestionSqlDialect.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlFailedMessageIngestionSqlDialect.cs index f1c9f0a0be..daf82efc03 100644 --- a/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlFailedMessageIngestionSqlDialect.cs +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlFailedMessageIngestionSqlDialect.cs @@ -63,6 +63,36 @@ ON CONFLICT (id) DO NOTHING } } + public async Task ResolveRetriedMessages(ServiceControlDbContext dbContext, IReadOnlyList 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,6 +126,11 @@ .. 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; @@ -103,8 +138,8 @@ static string BuildOnConflictUpdate() 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, diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerFailedMessageIngestionSqlDialect.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerFailedMessageIngestionSqlDialect.cs index cb8f4c9f53..89c3c5609a 100644 --- a/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerFailedMessageIngestionSqlDialect.cs +++ b/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 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,6 +140,11 @@ .. 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; @@ -118,8 +152,8 @@ static string BuildWhenMatchedUpdate() 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, diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFIngestionUnitOfWork.cs b/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFIngestionUnitOfWork.cs index acac97d01f..186c1e8790 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFIngestionUnitOfWork.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFIngestionUnitOfWork.cs @@ -17,7 +17,7 @@ public class EFIngestionUnitOfWork : IIngestionUnitOfWork readonly ConcurrentQueue failedProcessingAttempts = new(); readonly ConcurrentQueue bodyWrites = new(); readonly ConcurrentQueue knownEndpoints = new(); - readonly ConcurrentQueue confirmedRetries = new(); + readonly ConcurrentQueue 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) { diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFRecoverabilityIngestionUnitOfWork.cs b/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFRecoverabilityIngestionUnitOfWork.cs index 8566b94bcf..03739fbad0 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFRecoverabilityIngestionUnitOfWork.cs +++ b/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; } diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/FailedMessageBatchWriter.cs b/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/FailedMessageBatchWriter.cs index c451b0dd6b..0bd86675b8 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/FailedMessageBatchWriter.cs +++ b/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 attempts, IReadOnlyCollection knownEndpoints, - IReadOnlyCollection confirmedRetries, + IReadOnlyCollection 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 confirmedRetries) => + [.. confirmedRetries + .GroupBy(retry => retry.UniqueMessageId) + .Select(group => new ConfirmedRetry(group.Key, group.Max(retry => retry.SucceededAt))) + .OrderBy(retry => retry.UniqueMessageId)]; + static List BuildEndpointRows(IReadOnlyCollection 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); } } diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/ConfirmedRetry.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/ConfirmedRetry.cs new file mode 100644 index 0000000000..49da96daac --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/ConfirmedRetry.cs @@ -0,0 +1,7 @@ +namespace ServiceControl.Persistence.EFCore.Infrastructure; + +/// +/// A retry acknowledgement, carrying the time the retry succeeded so that a message which failed +/// again afterwards is not resolved by it. +/// +public readonly record struct ConfirmedRetry(Guid UniqueMessageId, DateTime SucceededAt); \ No newline at end of file diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/IFailedMessageIngestionSqlDialect.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/IFailedMessageIngestionSqlDialect.cs index 77bdd8aa19..8f7129ba57 100644 --- a/src/ServiceControl.Persistence.EFCore/Infrastructure/IFailedMessageIngestionSqlDialect.cs +++ b/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. /// Task InsertMissingKnownEndpoints(ServiceControlDbContext dbContext, IReadOnlyList rows, CancellationToken cancellationToken = default); + + /// + /// 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. + /// + Task ResolveRetriedMessages(ServiceControlDbContext dbContext, IReadOnlyList rows, DateTime now, CancellationToken cancellationToken = default); } diff --git a/src/ServiceControl.Persistence.RavenDB/ExpirationManager.cs b/src/ServiceControl.Persistence.RavenDB/ExpirationManager.cs index 39198bcc0d..bca8b10025 100644 --- a/src/ServiceControl.Persistence.RavenDB/ExpirationManager.cs +++ b/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'];"; } } \ No newline at end of file diff --git a/src/ServiceControl.Persistence.RavenDB/UnitOfWork/RavenRecoverabilityIngestionUnitOfWork.cs b/src/ServiceControl.Persistence.RavenDB/UnitOfWork/RavenRecoverabilityIngestionUnitOfWork.cs index b53113beea..e51458fd0c 100644 --- a/src/ServiceControl.Persistence.RavenDB/UnitOfWork/RavenRecoverabilityIngestionUnitOfWork.cs +++ b/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 { { "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 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; diff --git a/src/ServiceControl.Persistence.Tests.RavenDB/Expiration/MessageExpiryTests.cs b/src/ServiceControl.Persistence.Tests.RavenDB/Expiration/MessageExpiryTests.cs index 58d36c02d3..9b261b9a03 100644 --- a/src/ServiceControl.Persistence.Tests.RavenDB/Expiration/MessageExpiryTests.cs +++ b/src/ServiceControl.Persistence.Tests.RavenDB/Expiration/MessageExpiryTests.cs @@ -161,7 +161,7 @@ public async Task MessageFailingAgainAfterRetryShouldNotKeepExpiration() // Successful retry stamps @expires on the FailedMessage document. await using (var uow = await IngestionUnitOfWorkFactory.StartNew()) { - await uow.Recoverability.RecordSuccessfulRetry(uniqueMessageId); + await uow.Recoverability.RecordSuccessfulRetry(uniqueMessageId, attempt.AttemptedAt.AddMinutes(1)); await uow.Complete(TestContext.CurrentContext.CancellationToken); } @@ -169,7 +169,7 @@ public async Task MessageFailingAgainAfterRetryShouldNotKeepExpiration() await CompleteDatabaseOperation(); // The same logical message fails again before the retention period elapses. - var (context2, attempt2) = CreateMessageContext(uniqueMessageId); + var (context2, attempt2) = CreateMessageContext(uniqueMessageId, attempt.AttemptedAt.AddMinutes(2)); await using (var uow = await IngestionUnitOfWorkFactory.StartNew()) { @@ -211,7 +211,7 @@ public async Task RetryConfirmationProcessingShouldTriggerExpiration() await using (var uow = await IngestionUnitOfWorkFactory.StartNew()) { - await uow.Recoverability.RecordSuccessfulRetry(errors.Results.First().Id); + await uow.Recoverability.RecordSuccessfulRetry(errors.Results.First().Id, attempt.AttemptedAt.AddMinutes(1)); await uow.Complete(TestContext.CurrentContext.CancellationToken); } @@ -219,7 +219,7 @@ public async Task RetryConfirmationProcessingShouldTriggerExpiration() await WaitUntil(async () => (await GetAllMessages()).Results.Count == 0, "Retry confirmation should cause message removal."); } - static (MessageContext, FailedMessage.ProcessingAttempt) CreateMessageContext(string forceUniqueMessageId = null) + static (MessageContext, FailedMessage.ProcessingAttempt) CreateMessageContext(string forceUniqueMessageId = null, DateTime? attemptedAt = null) { var headers = new Dictionary { @@ -236,6 +236,11 @@ public async Task RetryConfirmationProcessingShouldTriggerExpiration() var attempt = FailedMessageBuilder.Minimal().ProcessingAttempts.First(); + if (attemptedAt.HasValue) + { + attempt.AttemptedAt = attemptedAt.Value; + } + var message = new MessageContext(Guid.NewGuid().ToString(), headers, ReadOnlyMemory.Empty, new TransportTransaction(), "receiveAddress", new ContextBag()); return (message, attempt); diff --git a/src/ServiceControl.Persistence.Tests.RavenDB/Recoverability/RetryConfirmationOrderingTests.cs b/src/ServiceControl.Persistence.Tests.RavenDB/Recoverability/RetryConfirmationOrderingTests.cs new file mode 100644 index 0000000000..d5581bd1ae --- /dev/null +++ b/src/ServiceControl.Persistence.Tests.RavenDB/Recoverability/RetryConfirmationOrderingTests.cs @@ -0,0 +1,79 @@ +namespace ServiceControl.Persistence.Tests.RavenDB.Recoverability; + +using System; +using System.Threading.Tasks; +using NUnit.Framework; +using ServiceControl.MessageFailures; + +// A retry acknowledgement and a later failure of the same message reach storage in whatever order +// their batches commit. Both orders are forced here rather than raced, so the assertion is about +// the end state and not about which batch got there first. +[TestFixture] +class RetryConfirmationOrderingTests : RavenPersistenceTestBase +{ + [Test] + public async Task A_confirmation_arriving_after_a_later_attempt_leaves_the_message_unresolved() + { + var failure = new IngestedFailure(); + var retrySucceededAt = failure.AttemptedAt.AddMinutes(1); + + await Ingest(failure); + await Ingest(failure.NextAttempt(retrySucceededAt.AddMinutes(1))); + await ConfirmRetry(failure.UniqueMessageIdString, retrySucceededAt); + + var message = await FailedMessageQueryStore.GetFailedMessage(failure.UniqueMessageIdString); + + Assert.That(message.Status, Is.EqualTo(FailedMessageStatus.Unresolved), "the message failed again after the retry succeeded"); + } + + [Test] + public async Task A_later_attempt_arriving_after_a_confirmation_leaves_the_message_unresolved() + { + var failure = new IngestedFailure(); + var retrySucceededAt = failure.AttemptedAt.AddMinutes(1); + + await Ingest(failure); + await ConfirmRetry(failure.UniqueMessageIdString, retrySucceededAt); + await Ingest(failure.NextAttempt(retrySucceededAt.AddMinutes(1))); + + var message = await FailedMessageQueryStore.GetFailedMessage(failure.UniqueMessageIdString); + + Assert.That(message.Status, Is.EqualTo(FailedMessageStatus.Unresolved), "the message failed again after the retry succeeded"); + } + + [Test] + public async Task A_redelivered_attempt_arriving_after_a_confirmation_leaves_the_message_resolved() + { + var failure = new IngestedFailure(); + + await Ingest(failure); + await ConfirmRetry(failure.UniqueMessageIdString, failure.AttemptedAt.AddMinutes(1)); + await Ingest(failure); + + var message = await FailedMessageQueryStore.GetFailedMessage(failure.UniqueMessageIdString); + + Assert.That(message.Status, Is.EqualTo(FailedMessageStatus.Resolved), "redelivering the attempt the retry was for is not a new failure"); + } + + async Task Ingest(IngestedFailure failure) + { + await using (var unitOfWork = await UnitOfWorkFactory.StartNew()) + { + await unitOfWork.Recoverability.RecordFailedProcessingAttempt(failure.Context, failure.ProcessingAttempt, failure.Groups); + await unitOfWork.Complete(TestContext.CurrentContext.CancellationToken); + } + + await CompleteDatabaseOperation(); + } + + async Task ConfirmRetry(string uniqueMessageId, DateTime succeededAt) + { + await using (var unitOfWork = await UnitOfWorkFactory.StartNew()) + { + await unitOfWork.Recoverability.RecordSuccessfulRetry(uniqueMessageId, succeededAt); + await unitOfWork.Complete(TestContext.CurrentContext.CancellationToken); + } + + await CompleteDatabaseOperation(); + } +} \ No newline at end of file diff --git a/src/ServiceControl.Persistence.Tests/EFCore/EditFailedMessagesDataStoreRetentionTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/EditFailedMessagesDataStoreRetentionTests.cs index 8ec279cb72..108a6caa18 100644 --- a/src/ServiceControl.Persistence.Tests/EFCore/EditFailedMessagesDataStoreRetentionTests.cs +++ b/src/ServiceControl.Persistence.Tests/EFCore/EditFailedMessagesDataStoreRetentionTests.cs @@ -18,8 +18,6 @@ class EditFailedMessagesDataStoreRetentionTests : ErrorIngestionTestBase { IEditFailedMessagesDataStore EditStore => ServiceProvider.GetRequiredService(); - DateTime Now => PersistenceTestsContext.FakeTime.GetUtcNow().UtcDateTime; - [Test] public async Task TryBeginEdit_stamps_StatusChangedAt_and_LastModified() { diff --git a/src/ServiceControl.Persistence.Tests/EFCore/ErrorIngestionConcurrencyTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/ErrorIngestionConcurrencyTests.cs index 4466341bbb..f785d68e7f 100644 --- a/src/ServiceControl.Persistence.Tests/EFCore/ErrorIngestionConcurrencyTests.cs +++ b/src/ServiceControl.Persistence.Tests/EFCore/ErrorIngestionConcurrencyTests.cs @@ -7,6 +7,7 @@ namespace ServiceControl.Persistence.Tests; using NUnit.Framework; using ServiceControl.MessageFailures; using ServiceControl.Operations; +using ServiceControl.Persistence.EFCore.Entities; class ErrorIngestionConcurrencyTests : ErrorIngestionTestBase { @@ -84,6 +85,75 @@ await Task.WhenAll(Enumerable.Range(0, writers).Select(writer => Task.Run(async } } + // A retry acknowledgement and a later failure of the same message reach storage in whatever + // order their batches commit, and with several ingestion hosts they need not even be on the + // same one. Both orders are forced here rather than raced, so the assertion is about the end + // state and not about which batch got there first. + [Test] + public async Task A_confirmation_arriving_after_a_later_attempt_leaves_the_message_unresolved() + { + var failure = new IngestedFailure(); + var retrySucceededAt = failure.AttemptedAt.AddMinutes(1); + + await Ingest(failure); + await Ingest(failure.NextAttempt(retrySucceededAt.AddMinutes(1))); + await ConfirmRetryAt(retrySucceededAt, failure.UniqueMessageIdString); + + var row = await GetFailedMessage(failure.UniqueMessageId); + + Assert.That(row.Status, Is.EqualTo(FailedMessageStatus.Unresolved), "the message failed again after the retry succeeded"); + } + + [Test] + public async Task A_later_attempt_arriving_after_a_confirmation_leaves_the_message_unresolved() + { + var failure = new IngestedFailure(); + var retrySucceededAt = failure.AttemptedAt.AddMinutes(1); + + await Ingest(failure); + await ConfirmRetryAt(retrySucceededAt, failure.UniqueMessageIdString); + await Ingest(failure.NextAttempt(retrySucceededAt.AddMinutes(1))); + + var row = await GetFailedMessage(failure.UniqueMessageId); + + Assert.That(row.Status, Is.EqualTo(FailedMessageStatus.Unresolved), "the message failed again after the retry succeeded"); + } + + [Test] + public async Task A_redelivered_attempt_arriving_after_a_confirmation_leaves_the_message_resolved() + { + var failure = new IngestedFailure(); + + await Ingest(failure); + await ConfirmRetryAt(failure.AttemptedAt.AddMinutes(1), failure.UniqueMessageIdString); + await Ingest(failure); + + var row = await GetFailedMessage(failure.UniqueMessageId); + + Assert.That(row.Status, Is.EqualTo(FailedMessageStatus.Resolved), "redelivering the attempt the retry was for is not a new failure"); + } + + [Test] + public async Task A_confirmation_releases_the_retry_claim_even_when_it_cannot_resolve() + { + var failure = new IngestedFailure(); + var retrySucceededAt = failure.AttemptedAt.AddMinutes(1); + + await Ingest(failure); + await Store(new FailedMessageRetryEntity { UniqueMessageId = failure.UniqueMessageId, RetryBatchId = Guid.NewGuid() }); + await Ingest(failure.NextAttempt(retrySucceededAt.AddMinutes(1))); + + await ConfirmRetryAt(retrySucceededAt, failure.UniqueMessageIdString); + + var row = await GetFailedMessage(failure.UniqueMessageId); + + using (Assert.EnterMultipleScope()) + { + Assert.That(row.Status, Is.EqualTo(FailedMessageStatus.Unresolved)); + Assert.That(await CountRetryRows(failure.UniqueMessageId), Is.Zero, "the retry itself completed, so its claim is released either way"); + } + } + [Test] public async Task Concurrent_writers_recording_the_same_endpoint_insert_it_once() { diff --git a/src/ServiceControl.Persistence.Tests/EFCore/ErrorIngestionTestBase.cs b/src/ServiceControl.Persistence.Tests/EFCore/ErrorIngestionTestBase.cs index e269c94761..fff36a40f9 100644 --- a/src/ServiceControl.Persistence.Tests/EFCore/ErrorIngestionTestBase.cs +++ b/src/ServiceControl.Persistence.Tests/EFCore/ErrorIngestionTestBase.cs @@ -43,12 +43,16 @@ protected Task Ingest(params IngestedFailure[] failures) => } }); - protected Task ConfirmRetry(params string[] uniqueMessageIds) => + protected DateTime Now => PersistenceTestsContext.FakeTime.GetUtcNow().UtcDateTime; + + protected Task ConfirmRetry(params string[] uniqueMessageIds) => ConfirmRetryAt(Now, uniqueMessageIds); + + protected Task ConfirmRetryAt(DateTime succeededAt, params string[] uniqueMessageIds) => InBatch(async unitOfWork => { foreach (var uniqueMessageId in uniqueMessageIds) { - await unitOfWork.Recoverability.RecordSuccessfulRetry(uniqueMessageId); + await unitOfWork.Recoverability.RecordSuccessfulRetry(uniqueMessageId, succeededAt); } }); diff --git a/src/ServiceControl.Persistence.Tests/EFCore/ErrorIngestionTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/ErrorIngestionTests.cs index fa3f1160d2..84510d54e5 100644 --- a/src/ServiceControl.Persistence.Tests/EFCore/ErrorIngestionTests.cs +++ b/src/ServiceControl.Persistence.Tests/EFCore/ErrorIngestionTests.cs @@ -263,7 +263,7 @@ public async Task A_failure_confirmed_in_the_same_batch_ends_resolved() await InBatch(async unitOfWork => { await unitOfWork.Recoverability.RecordFailedProcessingAttempt(failure.Context, failure.ProcessingAttempt, failure.Groups); - await unitOfWork.Recoverability.RecordSuccessfulRetry(failure.UniqueMessageIdString); + await unitOfWork.Recoverability.RecordSuccessfulRetry(failure.UniqueMessageIdString, Now); }); var row = await GetFailedMessage(failure.UniqueMessageId); diff --git a/src/ServiceControl.Persistence.Tests/EFCore/RetentionSweepTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/RetentionSweepTests.cs index 35aa3d930a..f556946af2 100644 --- a/src/ServiceControl.Persistence.Tests/EFCore/RetentionSweepTests.cs +++ b/src/ServiceControl.Persistence.Tests/EFCore/RetentionSweepTests.cs @@ -15,8 +15,6 @@ class RetentionSweepTests : ErrorIngestionTestBase [SetUp] public void SetRetention() => EFSettings.ErrorRetentionPeriod = TimeSpan.FromDays(30); - DateTime Now => PersistenceTestsContext.FakeTime.GetUtcNow().UtcDateTime; - [Test] public async Task Deletes_resolved_and_archived_rows_past_the_cutoff() { diff --git a/src/ServiceControl.Persistence/UnitOfWork/IRecoverabilityIngestionUnitOfWork.cs b/src/ServiceControl.Persistence/UnitOfWork/IRecoverabilityIngestionUnitOfWork.cs index 6ea5d43621..caf71a9a31 100644 --- a/src/ServiceControl.Persistence/UnitOfWork/IRecoverabilityIngestionUnitOfWork.cs +++ b/src/ServiceControl.Persistence/UnitOfWork/IRecoverabilityIngestionUnitOfWork.cs @@ -1,5 +1,6 @@ namespace ServiceControl.Persistence.UnitOfWork { + using System; using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; @@ -12,6 +13,12 @@ Task RecordFailedProcessingAttempt(MessageContext context, FailedMessage.ProcessingAttempt processingAttempt, List groups, CancellationToken cancellationToken = default); - Task RecordSuccessfulRetry(string retriedMessageUniqueId, CancellationToken cancellationToken = default); + /// + /// Resolves the message the retry was for, unless an attempt made after + /// has already been recorded. That attempt means the message + /// failed again after the retry succeeded, and the two can reach storage in either order, + /// from separate batches or from separate ingestion instances. + /// + Task RecordSuccessfulRetry(string retriedMessageUniqueId, DateTime succeededAt, CancellationToken cancellationToken = default); } } \ No newline at end of file diff --git a/src/ServiceControl/Operations/RetryConfirmationProcessor.cs b/src/ServiceControl/Operations/RetryConfirmationProcessor.cs index 8b948a7b91..79b653dff1 100644 --- a/src/ServiceControl/Operations/RetryConfirmationProcessor.cs +++ b/src/ServiceControl/Operations/RetryConfirmationProcessor.cs @@ -1,10 +1,12 @@ namespace ServiceControl.Operations { + using System; using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; using Contracts.MessageFailures; using Infrastructure.DomainEvents; + using NServiceBus; using NServiceBus.Transport; using ServiceControl.Persistence.UnitOfWork; @@ -23,7 +25,7 @@ public async Task Process(List contexts, IIngestionUnitOfWork un foreach (var context in contexts) { var retriedMessageUniqueId = context.Headers[RetryUniqueMessageIdHeader]; - await unitOfWork.Recoverability.RecordSuccessfulRetry(retriedMessageUniqueId, cancellationToken); + await unitOfWork.Recoverability.RecordSuccessfulRetry(retriedMessageUniqueId, GetSucceededAt(context.Headers), cancellationToken); } } @@ -35,6 +37,27 @@ public Task Announce(MessageContext messageContext, CancellationToken cancellati }, cancellationToken); } + // An acknowledgement that carries no readable time leaves nothing to order the confirmation + // against, so it is taken to be the most recent thing that happened to the message. That is + // how every confirmation was treated before the time was read at all. + static DateTime GetSucceededAt(Dictionary headers) + { + if (headers.TryGetValue(SuccessfulRetryHeader, out var wireFormattedTime) && !string.IsNullOrWhiteSpace(wireFormattedTime)) + { + try + { + return DateTimeOffsetHelper.ToDateTimeOffset(wireFormattedTime).UtcDateTime; + } + catch (FormatException) + { + } + } + + return NewerThanAnyAttempt; + } + + static readonly DateTime NewerThanAnyAttempt = DateTime.SpecifyKind(DateTime.MaxValue, DateTimeKind.Utc); + readonly IDomainEvents domainEvents; } } \ No newline at end of file