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