From c3c5ea5443a0725e4ce12e2a5f94413a2ffe08dd Mon Sep 17 00:00:00 2001 From: Rhys Bevilaqua Date: Tue, 11 Aug 2026 11:24:42 +0800 Subject: [PATCH 1/3] Implement retry failure store --- .../FailedMessageRetryDataStore.cs | 92 +++++- .../FailedMessageRetryDataStoreTests.cs | 278 ++++++++++++++++++ 2 files changed, 360 insertions(+), 10 deletions(-) create mode 100644 src/ServiceControl.Persistence.Tests/EFCore/FailedMessageRetryDataStoreTests.cs diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageRetryDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageRetryDataStore.cs index 12bc2c9c68..6cfbd9e666 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageRetryDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageRetryDataStore.cs @@ -1,18 +1,90 @@ namespace ServiceControl.Persistence.EFCore.Implementation; +using System.IO; +using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.DependencyInjection; +using ServiceControl.MessageFailures; +using ServiceControl.Operations.BodyStorage; +using ServiceControl.Persistence.EFCore.DbContexts; +using ServiceControl.Persistence.EFCore.Infrastructure; -public class FailedMessageRetryDataStore(IServiceScopeFactory scopeFactory) : DataStoreBase(scopeFactory), IFailedMessageRetryDataStore +public class FailedMessageRetryDataStore(IServiceScopeFactory scopeFactory, IBodyStorage bodyStorage) + : DataStoreBase(scopeFactory), IFailedMessageRetryDataStore { - public Task ProcessPendingRetries(DateTime periodFrom, DateTime periodTo, string queueAddress, Func processCallback, CancellationToken cancellationToken = default) => - throw new NotImplementedException(); + public Task RemoveFailedMessageRetry(string uniqueMessageId, CancellationToken cancellationToken) => + ExecuteWithDbContext(async dbContext => + { + if (!Guid.TryParse(uniqueMessageId, out var id)) + { + return; + } - public Task GetRetryPendingMessages(DateTime from, DateTime to, string queueAddress, CancellationToken cancellationToken = default) => - throw new NotImplementedException(); + // Point delete by primary key. ExecuteDeleteAsync returns the affected row count + // (0 when absent), so this is idempotent and never throws on a missing row — matching + // the RavenDB behaviour where a DeleteDocumentCommand on a non-existent doc is a no-op. + await dbContext.FailedMessageRetries + .Where(r => r.UniqueMessageId == id) + .ExecuteDeleteAsync(cancellationToken: cancellationToken); + }); - public Task RemoveFailedMessageRetry(string uniqueMessageId, CancellationToken cancellationToken = default) => - throw new NotImplementedException(); + public Task GetRetryPendingMessages(DateTime from, DateTime to, string queueAddress, CancellationToken cancellationToken) => + ExecuteWithDbContext(async dbContext => + { + // Pending retries are FailedMessages whose status is RetryIssued and whose LastModified + // falls within [from, to]. The RetryStagingStore.MarkBatchAsForwarding sets both + // LastModified and StatusChangedAt to "now" when a retry is issued, so LastModified + // reflects when the retry was sent — matching the RavenDB reference, which filters on + // LastModified for both pending-retry queries. + var normalizedQueueAddress = queueAddress?.ToLowerInvariant(); + var ids = await dbContext.FailedMessages + .AsNoTracking() + .Where(m => m.Status == FailedMessageStatus.RetryIssued + && m.LastModified >= from && m.LastModified <= to + && ((m.FailingEndpointAddress == null && normalizedQueueAddress == null) + || (m.FailingEndpointAddress != null && m.FailingEndpointAddress.ToLower() == normalizedQueueAddress))) + .Select(m => m.UniqueMessageId.ToString()) + .ToListAsync(cancellationToken: cancellationToken); - public Task GetFailedMessageBody(string uniqueMessageId, CancellationToken cancellationToken = default) => - throw new NotImplementedException(); -} + return ids.ToArray(); + }); + + public async Task ProcessPendingRetries(DateTime periodFrom, DateTime periodTo, string queueAddress, Func processCallback, CancellationToken cancellationToken) + { + // The callback is invoked while the DbContext scope is still open, so this method manages + // its own scope rather than using ExecuteWithDbContext (which disposes the scope before + // the caller can consume an AsAsyncEnumerable result). This mirrors how other streaming + // stores in the EFCore persister hold the scope open for the duration of enumeration. + await using var scope = scopeFactory.CreateAsyncScope(); + var dbContext = scope.ServiceProvider.GetRequiredService(); + + var query = dbContext.FailedMessages + .AsNoTracking() + .Where(m => m.Status == FailedMessageStatus.RetryIssued + && m.LastModified >= periodFrom && m.LastModified <= periodTo) + .FilterByQueueAddress(queueAddress) + .Select(m => m.UniqueMessageId.ToString()); + + await foreach (var uniqueMessageId in query.AsAsyncEnumerable().WithCancellation(cancellationToken)) + { + await processCallback(uniqueMessageId, cancellationToken); + } + } + + public async Task GetFailedMessageBody(string uniqueMessageId, CancellationToken cancellationToken) + { + var result = await bodyStorage.TryFetch(uniqueMessageId) + ?? throw new InvalidOperationException("IBodyStorage.TryFetch result cannot be null"); + + if (!result.HasResult) + { + throw new InvalidOperationException("IBodyStorage.TryFetch did not return a body"); + } + + await using (result.Stream) + { + using var memoryStream = new MemoryStream(); + await result.Stream.CopyToAsync(memoryStream, cancellationToken); + return memoryStream.ToArray(); + } + } +} \ No newline at end of file diff --git a/src/ServiceControl.Persistence.Tests/EFCore/FailedMessageRetryDataStoreTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/FailedMessageRetryDataStoreTests.cs new file mode 100644 index 0000000000..4936d8dd93 --- /dev/null +++ b/src/ServiceControl.Persistence.Tests/EFCore/FailedMessageRetryDataStoreTests.cs @@ -0,0 +1,278 @@ +namespace ServiceControl.Persistence.Tests; + +using System; +using System.Collections.Generic; +using System.Linq; +using System.Text; +using System.Threading.Tasks; +using NUnit.Framework; +using ServiceControl.MessageFailures; +using ServiceControl.Persistence.EFCore.Entities; + +class FailedMessageRetryDataStoreTests : ErrorIngestionTestBase +{ + static readonly DateTime Noon = new(2026, 8, 1, 12, 0, 0, DateTimeKind.Utc); + + [Test] + public async Task RemoveFailedMessageRetry_deletes_the_claim_row() + { + var id = await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon, queueAddress: "Shipping"); + await SeedRetryClaim(id); + + await FailedMessageRetryStore.RemoveFailedMessageRetry(id.ToString()); + + Assert.That(await CountRetryRows(id), Is.EqualTo(0), "the retry claim row should be deleted"); + } + + [Test] + public async Task RemoveFailedMessageRetry_is_idempotent_when_no_row_exists() + { + var id = await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon); + + // No claim row was seeded; the call must not throw. + Assert.DoesNotThrowAsync(() => FailedMessageRetryStore.RemoveFailedMessageRetry(id.ToString())); + Assert.That(await CountRetryRows(id), Is.EqualTo(0)); + } + + [Test] + public async Task RemoveFailedMessageRetry_ignores_a_non_guid_id() + { + // Consumers always pass a Guid string, but the store must not throw on garbage input. + Assert.DoesNotThrowAsync(() => FailedMessageRetryStore.RemoveFailedMessageRetry("not-a-guid")); + } + + [Test] + public async Task RemoveFailedMessageRetry_does_not_touch_the_failed_message_row() + { + var id = await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon, queueAddress: "Shipping"); + await SeedRetryClaim(id); + + await FailedMessageRetryStore.RemoveFailedMessageRetry(id.ToString()); + + var failed = await GetFailedMessage(id); + using (Assert.EnterMultipleScope()) + { + Assert.That(failed.Status, Is.EqualTo(FailedMessageStatus.RetryIssued), "status must be unchanged"); + Assert.That(failed.LastModified, Is.EqualTo(Noon), "LastModified must be unchanged"); + Assert.That(failed.FailingEndpointAddress, Is.EqualTo("Shipping")); + } + } + + [Test] + public async Task GetRetryPendingMessages_returns_only_RetryIssued_messages_in_the_window() + { + var inWindow = await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon, queueAddress: "Shipping"); + var beforeWindow = await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon.AddHours(-2)); + var afterWindow = await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon.AddHours(2)); + var unresolvedInWindow = await SeedFailedMessage(FailedMessageStatus.Unresolved, Noon, queueAddress: "Shipping"); + var resolvedInWindow = await SeedFailedMessage(FailedMessageStatus.Resolved, Noon, queueAddress: "Shipping"); + var archivedInWindow = await SeedFailedMessage(FailedMessageStatus.Archived, Noon, queueAddress: "Shipping"); + + var ids = await FailedMessageRetryStore.GetRetryPendingMessages(Noon.AddHours(-1), Noon.AddHours(1), "Shipping"); + + using (Assert.EnterMultipleScope()) + { + Assert.That(ids, Does.Contain(inWindow.ToString()).IgnoreCase, "the in-window RetryIssued message must be returned"); + Assert.That(ids, Does.Not.Contain(beforeWindow.ToString()).IgnoreCase, "messages before the window are excluded"); + Assert.That(ids, Does.Not.Contain(afterWindow.ToString()).IgnoreCase, "messages after the window are excluded"); + Assert.That(ids, Does.Not.Contain(unresolvedInWindow.ToString()).IgnoreCase, "Unresolved messages are excluded"); + Assert.That(ids, Does.Not.Contain(resolvedInWindow.ToString()).IgnoreCase, "Resolved messages are excluded"); + Assert.That(ids, Does.Not.Contain(archivedInWindow.ToString()).IgnoreCase, "Archived messages are excluded"); + } + } + + [Test] + public async Task GetRetryPendingMessages_returns_raw_guid_string_ids() + { + var id = await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon, queueAddress: "Shipping"); + + var ids = await FailedMessageRetryStore.GetRetryPendingMessages(Noon.AddHours(-1), Noon.AddHours(1), "Shipping"); + + Assert.That(ids, Has.Length.EqualTo(1)); + Assert.That(ids[0], Is.EqualTo(id.ToString()).IgnoreCase, "consumers Guid.Parse the returned id, so it must be a raw Guid string"); + Assert.That(Guid.TryParse(ids[0], out _), Is.True); + } + + [Test] + public async Task GetRetryPendingMessages_filters_by_queue_address() + { + var shipping = await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon, queueAddress: "Shipping"); + var billing = await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon, queueAddress: "Billing"); + + var ids = await FailedMessageRetryStore.GetRetryPendingMessages(Noon.AddHours(-1), Noon.AddHours(1), "Billing"); + + using (Assert.EnterMultipleScope()) + { + Assert.That(ids, Does.Contain(billing.ToString()).IgnoreCase); + Assert.That(ids, Does.Not.Contain(shipping.ToString()).IgnoreCase); + } + } + + [TestCase(null)] + [TestCase("")] + [TestCase(" ")] + public async Task GetRetryPendingMessages_with_a_null_or_blank_queue_address_does_not_match_other_queues(string queueAddress) + { + await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon, queueAddress: "Shipping"); + await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon, queueAddress: "Billing"); + + var ids = await FailedMessageRetryStore.GetRetryPendingMessages(Noon.AddHours(-1), Noon.AddHours(1), queueAddress); + + Assert.That(ids, Is.Empty); + } + + [Test] + public async Task ProcessPendingRetries_invokes_callback_once_per_matching_message_with_raw_guid_string() + { + var first = await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon, queueAddress: "Shipping"); + var second = await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon, queueAddress: "Shipping"); + var outOfWindow = await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon.AddHours(5), queueAddress: "Shipping"); + var unresolved = await SeedFailedMessage(FailedMessageStatus.Unresolved, Noon, queueAddress: "Shipping"); + + var captured = new List(); + var count = 0; + + await FailedMessageRetryStore.ProcessPendingRetries( + Noon.AddHours(-1), + Noon.AddHours(1), + "Shipping", + (id, _) => + { + captured.Add(id); + count++; + return Task.CompletedTask; + }); + + using (Assert.EnterMultipleScope()) + { + Assert.That(count, Is.EqualTo(2), "only the two in-window RetryIssued messages match"); + Assert.That(captured, Does.Contain(first.ToString()).IgnoreCase); + Assert.That(captured, Does.Contain(second.ToString()).IgnoreCase); + Assert.That(captured, Does.Not.Contain(outOfWindow.ToString()).IgnoreCase); + Assert.That(captured, Does.Not.Contain(unresolved.ToString()).IgnoreCase); + Assert.That(captured, Has.All.Matches(id => Guid.TryParse(id, out _)), "every callback id must be a raw Guid string"); + } + } + + [Test] + public async Task ProcessPendingRetries_with_null_queue_address_matches_all_queues() + { + var shipping = await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon, queueAddress: "Shipping"); + var billing = await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon, queueAddress: "Billing"); + + var captured = new List(); + + await FailedMessageRetryStore.ProcessPendingRetries( + Noon.AddHours(-1), + Noon.AddHours(1), + null, + (id, _) => + { + captured.Add(id); + return Task.CompletedTask; + }); + + using (Assert.EnterMultipleScope()) + { + Assert.That(captured, Does.Contain(shipping.ToString()).IgnoreCase); + Assert.That(captured, Does.Contain(billing.ToString()).IgnoreCase); + } + } + + [Test] + public async Task GetFailedMessageBody_returns_inline_BodyText_bytes() + { + var id = await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon, bodyText: "hello body"); + + var body = await FailedMessageRetryStore.GetFailedMessageBody(id.ToString()); + + Assert.That(body, Is.Not.Null, "an inline body must be returned"); + Assert.That(body, Is.EqualTo(Encoding.UTF8.GetBytes("hello body"))); + } + + [Test] + public void GetFailedMessageBody_throws_for_a_nonexistent_message() + { + Assert.ThrowsAsync(() => + FailedMessageRetryStore.GetFailedMessageBody(Guid.NewGuid().ToString())); + } + + [Test] + public async Task GetFailedMessageBody_throws_when_the_body_is_unavailable() + { + var id = await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon); + + Assert.ThrowsAsync(() => + FailedMessageRetryStore.GetFailedMessageBody(id.ToString())); + } + + [Test] + public async Task GetFailedMessageBody_returns_external_storage_body_when_BodyStoredExternally() + { + var id = await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon, bodyStoredExternally: true, bodyText: null); + var expected = Encoding.UTF8.GetBytes("external body payload"); + + await RecordedBodies.WriteBody(id.ToString(), expected, "text/plain"); + + var body = await FailedMessageRetryStore.GetFailedMessageBody(id.ToString()); + + Assert.That(body, Is.Not.Null); + Assert.That(body, Is.EqualTo(expected)); + } + + [Test] + public async Task RemoveFailedMessageRetry_removes_the_message_from_its_batch_staging_set() + { + var id = await SeedFailedMessage(FailedMessageStatus.Unresolved, Noon, queueAddress: "Shipping"); + var batchId = Guid.NewGuid(); + await SeedRetryClaim(id, batchId); + + // Before removal, the batch can stage the claimed message. + var before = await RetryStagingStore.GetMessagesToStage(batchId.ToString()); + Assert.That(before.Any(m => string.Equals(m.UniqueMessageId, id.ToString(), StringComparison.OrdinalIgnoreCase)), Is.True, "the claimed message should be staged by its batch"); + + await FailedMessageRetryStore.RemoveFailedMessageRetry(id.ToString()); + + var after = await RetryStagingStore.GetMessagesToStage(batchId.ToString()); + Assert.That(after.Any(m => string.Equals(m.UniqueMessageId, id.ToString(), StringComparison.OrdinalIgnoreCase)), Is.False, "deleting the claim row must drop the message from staging"); + } + + async Task SeedFailedMessage( + FailedMessageStatus status, + DateTime lastModified, + string queueAddress = null, + string bodyText = null, + bool bodyStoredExternally = false) + { + var id = Guid.NewGuid(); + + await Store(new FailedMessageEntity + { + UniqueMessageId = id, + Status = status, + StatusChangedAt = lastModified, + LastModified = lastModified, + NumberOfProcessingAttempts = 1, + FirstTimeOfFailure = lastModified, + LastTimeOfFailure = lastModified, + LastAttemptedAt = lastModified, + IsSystemMessage = false, + HeadersJson = "{}", + BodyText = bodyText, + BodyStoredExternally = bodyStoredExternally, + BodySize = bodyText is null ? 0 : Encoding.UTF8.GetByteCount(bodyText), + BodyContentType = bodyText is null ? null : "text/plain", + FailingEndpointAddress = queueAddress + }); + + return id; + } + + Task SeedRetryClaim(Guid uniqueMessageId, Guid? retryBatchId = null) => + Store(new FailedMessageRetryEntity + { + UniqueMessageId = uniqueMessageId, + RetryBatchId = retryBatchId ?? Guid.NewGuid(), + StageAttempts = 0 + }); +} \ No newline at end of file From 59cea8cf628979640c7176769ea653eab3846b0a Mon Sep 17 00:00:00 2001 From: Rhys Bevilaqua Date: Tue, 11 Aug 2026 12:36:40 +0800 Subject: [PATCH 2/3] Tidy up db call, make tests shared on all persisters --- .../FailedMessageRetryDataStore.cs | 35 +-- .../FailedMessageRetryBodyDataStoreTests.cs | 70 +++++ .../FailedMessageRetryDataStoreTests.cs | 278 ------------------ .../FailedMessageRetryDataStoreTests.cs | 175 +++++++++++ 4 files changed, 259 insertions(+), 299 deletions(-) create mode 100644 src/ServiceControl.Persistence.Tests/EFCore/FailedMessageRetryBodyDataStoreTests.cs delete mode 100644 src/ServiceControl.Persistence.Tests/EFCore/FailedMessageRetryDataStoreTests.cs create mode 100644 src/ServiceControl.Persistence.Tests/Recoverability/FailedMessageRetryDataStoreTests.cs diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageRetryDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageRetryDataStore.cs index 6cfbd9e666..5732300e67 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageRetryDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageRetryDataStore.cs @@ -5,7 +5,6 @@ namespace ServiceControl.Persistence.EFCore.Implementation; using Microsoft.Extensions.DependencyInjection; using ServiceControl.MessageFailures; using ServiceControl.Operations.BodyStorage; -using ServiceControl.Persistence.EFCore.DbContexts; using ServiceControl.Persistence.EFCore.Infrastructure; public class FailedMessageRetryDataStore(IServiceScopeFactory scopeFactory, IBodyStorage bodyStorage) @@ -48,27 +47,21 @@ public Task GetRetryPendingMessages(DateTime from, DateTime to, string return ids.ToArray(); }); - public async Task ProcessPendingRetries(DateTime periodFrom, DateTime periodTo, string queueAddress, Func processCallback, CancellationToken cancellationToken) - { - // The callback is invoked while the DbContext scope is still open, so this method manages - // its own scope rather than using ExecuteWithDbContext (which disposes the scope before - // the caller can consume an AsAsyncEnumerable result). This mirrors how other streaming - // stores in the EFCore persister hold the scope open for the duration of enumeration. - await using var scope = scopeFactory.CreateAsyncScope(); - var dbContext = scope.ServiceProvider.GetRequiredService(); - - var query = dbContext.FailedMessages - .AsNoTracking() - .Where(m => m.Status == FailedMessageStatus.RetryIssued - && m.LastModified >= periodFrom && m.LastModified <= periodTo) - .FilterByQueueAddress(queueAddress) - .Select(m => m.UniqueMessageId.ToString()); - - await foreach (var uniqueMessageId in query.AsAsyncEnumerable().WithCancellation(cancellationToken)) + public Task ProcessPendingRetries(DateTime periodFrom, DateTime periodTo, string queueAddress, Func processCallback, CancellationToken cancellationToken) => + ExecuteWithDbContext(async dbContext => { - await processCallback(uniqueMessageId, cancellationToken); - } - } + var query = dbContext.FailedMessages + .AsNoTracking() + .Where(m => m.Status == FailedMessageStatus.RetryIssued + && m.LastModified >= periodFrom && m.LastModified <= periodTo) + .FilterByQueueAddress(queueAddress) + .Select(m => m.UniqueMessageId.ToString()); + + await foreach (var uniqueMessageId in query.AsAsyncEnumerable().WithCancellation(cancellationToken)) + { + await processCallback(uniqueMessageId, cancellationToken); + } + }); public async Task GetFailedMessageBody(string uniqueMessageId, CancellationToken cancellationToken) { diff --git a/src/ServiceControl.Persistence.Tests/EFCore/FailedMessageRetryBodyDataStoreTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/FailedMessageRetryBodyDataStoreTests.cs new file mode 100644 index 0000000000..0a46b887cc --- /dev/null +++ b/src/ServiceControl.Persistence.Tests/EFCore/FailedMessageRetryBodyDataStoreTests.cs @@ -0,0 +1,70 @@ +namespace ServiceControl.Persistence.Tests; + +using System; +using System.Text; +using System.Threading.Tasks; +using NUnit.Framework; +using ServiceControl.MessageFailures; +using ServiceControl.Persistence.EFCore.Entities; + +class FailedMessageRetryBodyDataStoreTests : ErrorIngestionTestBase +{ + static readonly DateTime Noon = new(2026, 8, 1, 12, 0, 0, DateTimeKind.Utc); + + [Test] + public async Task GetFailedMessageBody_returns_inline_BodyText_bytes() + { + var id = await SeedFailedMessage(bodyText: "hello body"); + + var body = await FailedMessageRetryStore.GetFailedMessageBody(id.ToString()); + + Assert.That(body, Is.EqualTo(Encoding.UTF8.GetBytes("hello body"))); + } + + [Test] + public async Task GetFailedMessageBody_throws_when_the_body_is_unavailable() + { + var id = await SeedFailedMessage(); + + Assert.ThrowsAsync(() => + FailedMessageRetryStore.GetFailedMessageBody(id.ToString())); + } + + [Test] + public async Task GetFailedMessageBody_returns_external_storage_body_when_BodyStoredExternally() + { + var id = await SeedFailedMessage(bodyStoredExternally: true); + var expected = Encoding.UTF8.GetBytes("external body payload"); + + await RecordedBodies.WriteBody(id.ToString(), expected, "text/plain"); + + var body = await FailedMessageRetryStore.GetFailedMessageBody(id.ToString()); + + Assert.That(body, Is.EqualTo(expected)); + } + + async Task SeedFailedMessage(string bodyText = null, bool bodyStoredExternally = false) + { + var id = Guid.NewGuid(); + + await Store(new FailedMessageEntity + { + UniqueMessageId = id, + Status = FailedMessageStatus.RetryIssued, + StatusChangedAt = Noon, + LastModified = Noon, + NumberOfProcessingAttempts = 1, + FirstTimeOfFailure = Noon, + LastTimeOfFailure = Noon, + LastAttemptedAt = Noon, + IsSystemMessage = false, + HeadersJson = "{}", + BodyText = bodyText, + BodyStoredExternally = bodyStoredExternally, + BodySize = bodyText is null ? 0 : Encoding.UTF8.GetByteCount(bodyText), + BodyContentType = bodyText is null ? null : "text/plain" + }); + + return id; + } +} diff --git a/src/ServiceControl.Persistence.Tests/EFCore/FailedMessageRetryDataStoreTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/FailedMessageRetryDataStoreTests.cs deleted file mode 100644 index 4936d8dd93..0000000000 --- a/src/ServiceControl.Persistence.Tests/EFCore/FailedMessageRetryDataStoreTests.cs +++ /dev/null @@ -1,278 +0,0 @@ -namespace ServiceControl.Persistence.Tests; - -using System; -using System.Collections.Generic; -using System.Linq; -using System.Text; -using System.Threading.Tasks; -using NUnit.Framework; -using ServiceControl.MessageFailures; -using ServiceControl.Persistence.EFCore.Entities; - -class FailedMessageRetryDataStoreTests : ErrorIngestionTestBase -{ - static readonly DateTime Noon = new(2026, 8, 1, 12, 0, 0, DateTimeKind.Utc); - - [Test] - public async Task RemoveFailedMessageRetry_deletes_the_claim_row() - { - var id = await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon, queueAddress: "Shipping"); - await SeedRetryClaim(id); - - await FailedMessageRetryStore.RemoveFailedMessageRetry(id.ToString()); - - Assert.That(await CountRetryRows(id), Is.EqualTo(0), "the retry claim row should be deleted"); - } - - [Test] - public async Task RemoveFailedMessageRetry_is_idempotent_when_no_row_exists() - { - var id = await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon); - - // No claim row was seeded; the call must not throw. - Assert.DoesNotThrowAsync(() => FailedMessageRetryStore.RemoveFailedMessageRetry(id.ToString())); - Assert.That(await CountRetryRows(id), Is.EqualTo(0)); - } - - [Test] - public async Task RemoveFailedMessageRetry_ignores_a_non_guid_id() - { - // Consumers always pass a Guid string, but the store must not throw on garbage input. - Assert.DoesNotThrowAsync(() => FailedMessageRetryStore.RemoveFailedMessageRetry("not-a-guid")); - } - - [Test] - public async Task RemoveFailedMessageRetry_does_not_touch_the_failed_message_row() - { - var id = await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon, queueAddress: "Shipping"); - await SeedRetryClaim(id); - - await FailedMessageRetryStore.RemoveFailedMessageRetry(id.ToString()); - - var failed = await GetFailedMessage(id); - using (Assert.EnterMultipleScope()) - { - Assert.That(failed.Status, Is.EqualTo(FailedMessageStatus.RetryIssued), "status must be unchanged"); - Assert.That(failed.LastModified, Is.EqualTo(Noon), "LastModified must be unchanged"); - Assert.That(failed.FailingEndpointAddress, Is.EqualTo("Shipping")); - } - } - - [Test] - public async Task GetRetryPendingMessages_returns_only_RetryIssued_messages_in_the_window() - { - var inWindow = await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon, queueAddress: "Shipping"); - var beforeWindow = await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon.AddHours(-2)); - var afterWindow = await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon.AddHours(2)); - var unresolvedInWindow = await SeedFailedMessage(FailedMessageStatus.Unresolved, Noon, queueAddress: "Shipping"); - var resolvedInWindow = await SeedFailedMessage(FailedMessageStatus.Resolved, Noon, queueAddress: "Shipping"); - var archivedInWindow = await SeedFailedMessage(FailedMessageStatus.Archived, Noon, queueAddress: "Shipping"); - - var ids = await FailedMessageRetryStore.GetRetryPendingMessages(Noon.AddHours(-1), Noon.AddHours(1), "Shipping"); - - using (Assert.EnterMultipleScope()) - { - Assert.That(ids, Does.Contain(inWindow.ToString()).IgnoreCase, "the in-window RetryIssued message must be returned"); - Assert.That(ids, Does.Not.Contain(beforeWindow.ToString()).IgnoreCase, "messages before the window are excluded"); - Assert.That(ids, Does.Not.Contain(afterWindow.ToString()).IgnoreCase, "messages after the window are excluded"); - Assert.That(ids, Does.Not.Contain(unresolvedInWindow.ToString()).IgnoreCase, "Unresolved messages are excluded"); - Assert.That(ids, Does.Not.Contain(resolvedInWindow.ToString()).IgnoreCase, "Resolved messages are excluded"); - Assert.That(ids, Does.Not.Contain(archivedInWindow.ToString()).IgnoreCase, "Archived messages are excluded"); - } - } - - [Test] - public async Task GetRetryPendingMessages_returns_raw_guid_string_ids() - { - var id = await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon, queueAddress: "Shipping"); - - var ids = await FailedMessageRetryStore.GetRetryPendingMessages(Noon.AddHours(-1), Noon.AddHours(1), "Shipping"); - - Assert.That(ids, Has.Length.EqualTo(1)); - Assert.That(ids[0], Is.EqualTo(id.ToString()).IgnoreCase, "consumers Guid.Parse the returned id, so it must be a raw Guid string"); - Assert.That(Guid.TryParse(ids[0], out _), Is.True); - } - - [Test] - public async Task GetRetryPendingMessages_filters_by_queue_address() - { - var shipping = await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon, queueAddress: "Shipping"); - var billing = await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon, queueAddress: "Billing"); - - var ids = await FailedMessageRetryStore.GetRetryPendingMessages(Noon.AddHours(-1), Noon.AddHours(1), "Billing"); - - using (Assert.EnterMultipleScope()) - { - Assert.That(ids, Does.Contain(billing.ToString()).IgnoreCase); - Assert.That(ids, Does.Not.Contain(shipping.ToString()).IgnoreCase); - } - } - - [TestCase(null)] - [TestCase("")] - [TestCase(" ")] - public async Task GetRetryPendingMessages_with_a_null_or_blank_queue_address_does_not_match_other_queues(string queueAddress) - { - await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon, queueAddress: "Shipping"); - await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon, queueAddress: "Billing"); - - var ids = await FailedMessageRetryStore.GetRetryPendingMessages(Noon.AddHours(-1), Noon.AddHours(1), queueAddress); - - Assert.That(ids, Is.Empty); - } - - [Test] - public async Task ProcessPendingRetries_invokes_callback_once_per_matching_message_with_raw_guid_string() - { - var first = await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon, queueAddress: "Shipping"); - var second = await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon, queueAddress: "Shipping"); - var outOfWindow = await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon.AddHours(5), queueAddress: "Shipping"); - var unresolved = await SeedFailedMessage(FailedMessageStatus.Unresolved, Noon, queueAddress: "Shipping"); - - var captured = new List(); - var count = 0; - - await FailedMessageRetryStore.ProcessPendingRetries( - Noon.AddHours(-1), - Noon.AddHours(1), - "Shipping", - (id, _) => - { - captured.Add(id); - count++; - return Task.CompletedTask; - }); - - using (Assert.EnterMultipleScope()) - { - Assert.That(count, Is.EqualTo(2), "only the two in-window RetryIssued messages match"); - Assert.That(captured, Does.Contain(first.ToString()).IgnoreCase); - Assert.That(captured, Does.Contain(second.ToString()).IgnoreCase); - Assert.That(captured, Does.Not.Contain(outOfWindow.ToString()).IgnoreCase); - Assert.That(captured, Does.Not.Contain(unresolved.ToString()).IgnoreCase); - Assert.That(captured, Has.All.Matches(id => Guid.TryParse(id, out _)), "every callback id must be a raw Guid string"); - } - } - - [Test] - public async Task ProcessPendingRetries_with_null_queue_address_matches_all_queues() - { - var shipping = await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon, queueAddress: "Shipping"); - var billing = await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon, queueAddress: "Billing"); - - var captured = new List(); - - await FailedMessageRetryStore.ProcessPendingRetries( - Noon.AddHours(-1), - Noon.AddHours(1), - null, - (id, _) => - { - captured.Add(id); - return Task.CompletedTask; - }); - - using (Assert.EnterMultipleScope()) - { - Assert.That(captured, Does.Contain(shipping.ToString()).IgnoreCase); - Assert.That(captured, Does.Contain(billing.ToString()).IgnoreCase); - } - } - - [Test] - public async Task GetFailedMessageBody_returns_inline_BodyText_bytes() - { - var id = await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon, bodyText: "hello body"); - - var body = await FailedMessageRetryStore.GetFailedMessageBody(id.ToString()); - - Assert.That(body, Is.Not.Null, "an inline body must be returned"); - Assert.That(body, Is.EqualTo(Encoding.UTF8.GetBytes("hello body"))); - } - - [Test] - public void GetFailedMessageBody_throws_for_a_nonexistent_message() - { - Assert.ThrowsAsync(() => - FailedMessageRetryStore.GetFailedMessageBody(Guid.NewGuid().ToString())); - } - - [Test] - public async Task GetFailedMessageBody_throws_when_the_body_is_unavailable() - { - var id = await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon); - - Assert.ThrowsAsync(() => - FailedMessageRetryStore.GetFailedMessageBody(id.ToString())); - } - - [Test] - public async Task GetFailedMessageBody_returns_external_storage_body_when_BodyStoredExternally() - { - var id = await SeedFailedMessage(FailedMessageStatus.RetryIssued, Noon, bodyStoredExternally: true, bodyText: null); - var expected = Encoding.UTF8.GetBytes("external body payload"); - - await RecordedBodies.WriteBody(id.ToString(), expected, "text/plain"); - - var body = await FailedMessageRetryStore.GetFailedMessageBody(id.ToString()); - - Assert.That(body, Is.Not.Null); - Assert.That(body, Is.EqualTo(expected)); - } - - [Test] - public async Task RemoveFailedMessageRetry_removes_the_message_from_its_batch_staging_set() - { - var id = await SeedFailedMessage(FailedMessageStatus.Unresolved, Noon, queueAddress: "Shipping"); - var batchId = Guid.NewGuid(); - await SeedRetryClaim(id, batchId); - - // Before removal, the batch can stage the claimed message. - var before = await RetryStagingStore.GetMessagesToStage(batchId.ToString()); - Assert.That(before.Any(m => string.Equals(m.UniqueMessageId, id.ToString(), StringComparison.OrdinalIgnoreCase)), Is.True, "the claimed message should be staged by its batch"); - - await FailedMessageRetryStore.RemoveFailedMessageRetry(id.ToString()); - - var after = await RetryStagingStore.GetMessagesToStage(batchId.ToString()); - Assert.That(after.Any(m => string.Equals(m.UniqueMessageId, id.ToString(), StringComparison.OrdinalIgnoreCase)), Is.False, "deleting the claim row must drop the message from staging"); - } - - async Task SeedFailedMessage( - FailedMessageStatus status, - DateTime lastModified, - string queueAddress = null, - string bodyText = null, - bool bodyStoredExternally = false) - { - var id = Guid.NewGuid(); - - await Store(new FailedMessageEntity - { - UniqueMessageId = id, - Status = status, - StatusChangedAt = lastModified, - LastModified = lastModified, - NumberOfProcessingAttempts = 1, - FirstTimeOfFailure = lastModified, - LastTimeOfFailure = lastModified, - LastAttemptedAt = lastModified, - IsSystemMessage = false, - HeadersJson = "{}", - BodyText = bodyText, - BodyStoredExternally = bodyStoredExternally, - BodySize = bodyText is null ? 0 : Encoding.UTF8.GetByteCount(bodyText), - BodyContentType = bodyText is null ? null : "text/plain", - FailingEndpointAddress = queueAddress - }); - - return id; - } - - Task SeedRetryClaim(Guid uniqueMessageId, Guid? retryBatchId = null) => - Store(new FailedMessageRetryEntity - { - UniqueMessageId = uniqueMessageId, - RetryBatchId = retryBatchId ?? Guid.NewGuid(), - StageAttempts = 0 - }); -} \ No newline at end of file diff --git a/src/ServiceControl.Persistence.Tests/Recoverability/FailedMessageRetryDataStoreTests.cs b/src/ServiceControl.Persistence.Tests/Recoverability/FailedMessageRetryDataStoreTests.cs new file mode 100644 index 0000000000..ea78b02c74 --- /dev/null +++ b/src/ServiceControl.Persistence.Tests/Recoverability/FailedMessageRetryDataStoreTests.cs @@ -0,0 +1,175 @@ +namespace ServiceControl.Persistence.Tests; + +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading.Tasks; +using NUnit.Framework; +using ServiceControl.MessageFailures; +using ServiceControl.Recoverability; + +class FailedMessageRetryDataStoreTests : PersistenceTestBase +{ + [Test] + public async Task RemoveFailedMessageRetry_removes_the_claim_without_touching_the_failed_message() + { + var failure = new IngestedFailure { FailingEndpointAddress = "Shipping" }; + var id = await Insert(failure); + var batchId = await StageBatch(id); + + Assert.That((await RetryStagingStore.GetMessagesToStage(batchId)).Select(m => m.UniqueMessageId), Does.Contain(id)); + + await FailedMessageRetryStore.RemoveFailedMessageRetry(id); + await CompleteDatabaseOperation(); + + var failedMessage = await FailedMessageQueryStore.GetFailedMessage(id); + using (Assert.EnterMultipleScope()) + { + Assert.That(await RetryStagingStore.GetMessagesToStage(batchId), Is.Empty, "the retry claim should be removed"); + Assert.That(failedMessage, Is.Not.Null, "the failed message should remain"); + Assert.That(failedMessage.Status, Is.EqualTo(FailedMessageStatus.Unresolved), "the failed-message status should be unchanged"); + } + } + + [Test] + public async Task RemoveFailedMessageRetry_is_idempotent_when_no_claim_exists() + { + var id = await Insert(new IngestedFailure()); + + Assert.DoesNotThrowAsync(() => FailedMessageRetryStore.RemoveFailedMessageRetry(id)); + Assert.That(await FailedMessageQueryStore.GetFailedMessage(id), Is.Not.Null, "the failed message should remain"); + } + + [Test] + public void RemoveFailedMessageRetry_ignores_a_non_guid_id() + { + Assert.DoesNotThrowAsync(() => FailedMessageRetryStore.RemoveFailedMessageRetry("not-a-guid")); + } + + [Test] + public async Task GetRetryPendingMessages_returns_only_matching_RetryIssued_messages_as_raw_ids() + { + var shipping = await Insert(new IngestedFailure { FailingEndpointAddress = "Shipping" }, FailedMessageStatus.RetryIssued); + var billing = await Insert(new IngestedFailure { FailingEndpointAddress = "Billing" }, FailedMessageStatus.RetryIssued); + var unresolved = await Insert(new IngestedFailure { FailingEndpointAddress = "Shipping" }, FailedMessageStatus.Unresolved); + var resolved = await Insert(new IngestedFailure { FailingEndpointAddress = "Shipping" }, FailedMessageStatus.Resolved); + var archived = await Insert(new IngestedFailure { FailingEndpointAddress = "Shipping" }, FailedMessageStatus.Archived); + + var ids = await FailedMessageRetryStore.GetRetryPendingMessages(WindowStart, WindowEnd, "Shipping"); + + using (Assert.EnterMultipleScope()) + { + Assert.That(ids, Is.EquivalentTo(new[] { shipping }).IgnoreCase); + Assert.That(ids, Does.Not.Contain(billing).IgnoreCase); + Assert.That(ids, Does.Not.Contain(unresolved).IgnoreCase); + Assert.That(ids, Does.Not.Contain(resolved).IgnoreCase); + Assert.That(ids, Does.Not.Contain(archived).IgnoreCase); + Assert.That(ids, Has.All.Matches(id => Guid.TryParse(id, out _)), "consumers expect raw unique-message IDs"); + } + } + + [Test] + public async Task GetRetryPendingMessages_excludes_messages_outside_the_window() + { + await Insert(new IngestedFailure { FailingEndpointAddress = "Shipping" }, FailedMessageStatus.RetryIssued); + + var ids = await FailedMessageRetryStore.GetRetryPendingMessages(DateTime.UtcNow.AddHours(1), DateTime.UtcNow.AddHours(2), "Shipping"); + + Assert.That(ids, Is.Empty); + } + + [TestCase(null)] + [TestCase("")] + [TestCase(" ")] + public async Task GetRetryPendingMessages_with_a_null_or_blank_queue_does_not_match_other_queues(string queueAddress) + { + await Insert(new IngestedFailure { FailingEndpointAddress = "Shipping" }, FailedMessageStatus.RetryIssued); + await Insert(new IngestedFailure { FailingEndpointAddress = "Billing" }, FailedMessageStatus.RetryIssued); + + var ids = await FailedMessageRetryStore.GetRetryPendingMessages(WindowStart, WindowEnd, queueAddress); + + Assert.That(ids, Is.Empty); + } + + [Test] + public async Task ProcessPendingRetries_invokes_the_callback_for_each_matching_message() + { + var first = await Insert(new IngestedFailure { FailingEndpointAddress = "Shipping" }, FailedMessageStatus.RetryIssued); + var second = await Insert(new IngestedFailure { FailingEndpointAddress = "Shipping" }, FailedMessageStatus.RetryIssued); + var billing = await Insert(new IngestedFailure { FailingEndpointAddress = "Billing" }, FailedMessageStatus.RetryIssued); + var unresolved = await Insert(new IngestedFailure { FailingEndpointAddress = "Shipping" }, FailedMessageStatus.Unresolved); + var captured = new List(); + + await FailedMessageRetryStore.ProcessPendingRetries(WindowStart, WindowEnd, "Shipping", (id, _) => + { + captured.Add(id); + return Task.CompletedTask; + }); + + using (Assert.EnterMultipleScope()) + { + Assert.That(captured, Is.EquivalentTo([first, second]).IgnoreCase); + Assert.That(captured, Does.Not.Contain(billing).IgnoreCase); + Assert.That(captured, Does.Not.Contain(unresolved).IgnoreCase); + Assert.That(captured, Has.All.Matches(id => Guid.TryParse(id, out _)), "callbacks should receive raw unique-message IDs"); + } + } + + [Test] + public async Task ProcessPendingRetries_with_a_null_queue_matches_all_queues() + { + var shipping = await Insert(new IngestedFailure { FailingEndpointAddress = "Shipping" }, FailedMessageStatus.RetryIssued); + var billing = await Insert(new IngestedFailure { FailingEndpointAddress = "Billing" }, FailedMessageStatus.RetryIssued); + var captured = new List(); + + await FailedMessageRetryStore.ProcessPendingRetries(WindowStart, WindowEnd, null, (id, _) => + { + captured.Add(id); + return Task.CompletedTask; + }); + + Assert.That(captured, Is.EquivalentTo(new[] { shipping, billing }).IgnoreCase); + } + + [Test] + public void GetFailedMessageBody_throws_for_a_nonexistent_message() + { + Assert.ThrowsAsync(() => + FailedMessageRetryStore.GetFailedMessageBody(Guid.NewGuid().ToString())); + } + + async Task Insert(IngestedFailure failure, FailedMessageStatus status = FailedMessageStatus.Unresolved) + { + var message = failure.ToFailedMessage(status); + message.Id = PersistenceTestsContext.GenerateFailedMessageRecordId(message.UniqueMessageId); + + await PersistenceTestsContext.InsertFailedMessages(message); + await CompleteDatabaseOperation(); + + return failure.UniqueMessageIdString; + } + + async Task StageBatch(params string[] uniqueMessageIds) + { + var batchId = await RetryBatchStore.CreateBatch( + RetryDocumentManager.RetrySessionId, + "request-1", + RetryType.MultipleMessages, + uniqueMessageIds, + "a retry request", + DateTime.UtcNow, + batchName: "a batch", + initiatedById: "alice-sub", + initiatedByName: "Alice", + operationId: "operation-1"); + + await RetryBatchStore.AssignMessagesToBatch(batchId, uniqueMessageIds); + await RetryBatchStore.MoveBatchToStaging(batchId); + await CompleteDatabaseOperation(); + + return batchId; + } + + static DateTime WindowStart => DateTime.UtcNow.AddMinutes(-10); + static DateTime WindowEnd => DateTime.UtcNow.AddMinutes(10); +} From 140f024d99a5c22d0ccdc1e350367484afff2bf1 Mon Sep 17 00:00:00 2001 From: Rhys Bevilaqua Date: Wed, 12 Aug 2026 09:33:29 +0800 Subject: [PATCH 3/3] Remove overly verbose comments --- .../FailedMessageRetryDataStore.cs | 24 ++++++------------- 1 file changed, 7 insertions(+), 17 deletions(-) diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageRetryDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageRetryDataStore.cs index 5732300e67..44f928c403 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageRetryDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageRetryDataStore.cs @@ -13,34 +13,24 @@ public class FailedMessageRetryDataStore(IServiceScopeFactory scopeFactory, IBod public Task RemoveFailedMessageRetry(string uniqueMessageId, CancellationToken cancellationToken) => ExecuteWithDbContext(async dbContext => { - if (!Guid.TryParse(uniqueMessageId, out var id)) + if (Guid.TryParse(uniqueMessageId, out var id)) { - return; + await dbContext.FailedMessageRetries + .Where(r => r.UniqueMessageId == id) + .ExecuteDeleteAsync(cancellationToken: cancellationToken); } - - // Point delete by primary key. ExecuteDeleteAsync returns the affected row count - // (0 when absent), so this is idempotent and never throws on a missing row — matching - // the RavenDB behaviour where a DeleteDocumentCommand on a non-existent doc is a no-op. - await dbContext.FailedMessageRetries - .Where(r => r.UniqueMessageId == id) - .ExecuteDeleteAsync(cancellationToken: cancellationToken); }); public Task GetRetryPendingMessages(DateTime from, DateTime to, string queueAddress, CancellationToken cancellationToken) => ExecuteWithDbContext(async dbContext => { - // Pending retries are FailedMessages whose status is RetryIssued and whose LastModified - // falls within [from, to]. The RetryStagingStore.MarkBatchAsForwarding sets both - // LastModified and StatusChangedAt to "now" when a retry is issued, so LastModified - // reflects when the retry was sent — matching the RavenDB reference, which filters on - // LastModified for both pending-retry queries. var normalizedQueueAddress = queueAddress?.ToLowerInvariant(); var ids = await dbContext.FailedMessages .AsNoTracking() .Where(m => m.Status == FailedMessageStatus.RetryIssued - && m.LastModified >= from && m.LastModified <= to - && ((m.FailingEndpointAddress == null && normalizedQueueAddress == null) - || (m.FailingEndpointAddress != null && m.FailingEndpointAddress.ToLower() == normalizedQueueAddress))) + && m.LastModified >= from && m.LastModified <= to + && ((m.FailingEndpointAddress == null && normalizedQueueAddress == null) + || (m.FailingEndpointAddress != null && m.FailingEndpointAddress.ToLower() == normalizedQueueAddress))) .Select(m => m.UniqueMessageId.ToString()) .ToListAsync(cancellationToken: cancellationToken);