diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageRetryDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageRetryDataStore.cs index 12bc2c9c68..44f928c403 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageRetryDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageRetryDataStore.cs @@ -1,18 +1,73 @@ 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.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)) + { + await dbContext.FailedMessageRetries + .Where(r => r.UniqueMessageId == id) + .ExecuteDeleteAsync(cancellationToken: cancellationToken); + } + }); - public Task GetRetryPendingMessages(DateTime from, DateTime to, string queueAddress, CancellationToken cancellationToken = default) => - throw new NotImplementedException(); + public Task GetRetryPendingMessages(DateTime from, DateTime to, string queueAddress, CancellationToken cancellationToken) => + ExecuteWithDbContext(async dbContext => + { + 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 RemoveFailedMessageRetry(string uniqueMessageId, CancellationToken cancellationToken = default) => - throw new NotImplementedException(); + return ids.ToArray(); + }); - public Task GetFailedMessageBody(string uniqueMessageId, CancellationToken cancellationToken = default) => - throw new NotImplementedException(); -} + public Task ProcessPendingRetries(DateTime periodFrom, DateTime periodTo, string queueAddress, Func processCallback, CancellationToken cancellationToken) => + ExecuteWithDbContext(async dbContext => + { + 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/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/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); +}