diff --git a/src/Directory.Packages.props b/src/Directory.Packages.props index 87dde27202..f866685c01 100644 --- a/src/Directory.Packages.props +++ b/src/Directory.Packages.props @@ -64,11 +64,11 @@ - + - + diff --git a/src/ServiceControl.AcceptanceTesting/NServiceBusAcceptanceTest.cs b/src/ServiceControl.AcceptanceTesting/NServiceBusAcceptanceTest.cs index 35a82c7ece..21fe3d4287 100644 --- a/src/ServiceControl.AcceptanceTesting/NServiceBusAcceptanceTest.cs +++ b/src/ServiceControl.AcceptanceTesting/NServiceBusAcceptanceTest.cs @@ -1,10 +1,13 @@ namespace ServiceControl.AcceptanceTesting { + using System; using System.Linq; using System.Threading; + using NServiceBus.AcceptanceTesting; using NServiceBus.AcceptanceTesting.Customization; using NServiceBus.Logging; using NUnit.Framework; + using NUnit.Framework.Internal; /// /// Base class for all the NSB test that sets up our conventions @@ -35,5 +38,22 @@ public void SetUp() return testName + "." + endpointBuilder; }; } + + [TearDown] + public void TearDown() + { + if (!TestExecutionContext.CurrentContext.TryGetRunDescriptor(out var runDescriptor)) + { + return; + } + + var scenarioContext = runDescriptor.ScenarioContext; + + TestContext.Out.WriteLine($@"Test settings: +{string.Join(Environment.NewLine, runDescriptor.Settings.Select(setting => $" {setting.Key}: {setting.Value}"))}"); + + TestContext.Out.WriteLine($@"Context: +{string.Join(Environment.NewLine, scenarioContext.GetType().GetProperties().Select(p => $"{p.Name} = {p.GetValue(scenarioContext, null)}"))}"); + } } } \ No newline at end of file diff --git a/src/ServiceControl.AcceptanceTests.RavenDB/ServiceControl.AcceptanceTests.RavenDB.csproj b/src/ServiceControl.AcceptanceTests.RavenDB/ServiceControl.AcceptanceTests.RavenDB.csproj index 602a756222..3dc48639d3 100644 --- a/src/ServiceControl.AcceptanceTests.RavenDB/ServiceControl.AcceptanceTests.RavenDB.csproj +++ b/src/ServiceControl.AcceptanceTests.RavenDB/ServiceControl.AcceptanceTests.RavenDB.csproj @@ -30,4 +30,8 @@ + + + + diff --git a/src/ServiceControl.AcceptanceTests/Recoverability/ExternalIntegration/FailedMessageExtensions.cs b/src/ServiceControl.AcceptanceTests/Recoverability/ExternalIntegration/FailedMessageExtensions.cs new file mode 100644 index 0000000000..262991a3b3 --- /dev/null +++ b/src/ServiceControl.AcceptanceTests/Recoverability/ExternalIntegration/FailedMessageExtensions.cs @@ -0,0 +1,26 @@ +using System.Collections.Generic; +using System.Linq; +using System.Threading.Tasks; +using ServiceControl.AcceptanceTesting; +using ServiceControl.AcceptanceTests; +using ServiceControl.MessageFailures.Api; + +public static class FailedMessageExtensions +{ + internal static async Task GetOnlyFailedUnresolvedMessageId(this AcceptanceTest test) + { + var allFailedMessages = + await test.TryGet>($"/api/errors/?status=unresolved"); + if (!allFailedMessages.HasResult) + { + return null; + } + + if (allFailedMessages.Item.Count != 1) + { + return null; + } + + return allFailedMessages.Item.First().Id; + } +} \ No newline at end of file diff --git a/src/ServiceControl.AcceptanceTests/Recoverability/ExternalIntegration/When_a_failed_edit_is_resolved_by_retry.cs b/src/ServiceControl.AcceptanceTests/Recoverability/ExternalIntegration/When_a_failed_edit_is_resolved_by_retry.cs new file mode 100644 index 0000000000..6441e3a5b1 --- /dev/null +++ b/src/ServiceControl.AcceptanceTests/Recoverability/ExternalIntegration/When_a_failed_edit_is_resolved_by_retry.cs @@ -0,0 +1,191 @@ +namespace ServiceControl.AcceptanceTests.Recoverability.ExternalIntegration +{ + using System.Collections.Generic; + using System.Linq; + using System.Threading.Tasks; + using AcceptanceTesting; + using AcceptanceTesting.EndpointTemplates; + using Contracts; + using NServiceBus; + using NServiceBus.AcceptanceTesting; + using NUnit.Framework; + using ServiceControl.MessageFailures; + using ServiceControl.MessageFailures.Api; + using JsonSerializer = System.Text.Json.JsonSerializer; + + class When_a_failed_edit_is_resolved_by_retry : AcceptanceTest + { + [Test] + public async Task Should_publish_notification() + { + CustomConfiguration = config => config.OnEndpointSubscribed((s, ctx) => + { + ctx.ExternalProcessorSubscribed = s.SubscriberReturnAddress.Contains(nameof(MessageReceiver)); + }); + + var context = await Define() + .WithEndpoint(b => b.When(async (bus, c) => + { + await bus.Subscribe(); + await bus.Subscribe(); + await bus.Subscribe(); + + if (c.HasNativePubSubSupport) + { + c.ExternalProcessorSubscribed = true; + } + }).When(c => c.SendLocal(new EditResolutionMessage())).DoNotFailOnErrorMessages()) + .Done(async ctx => + { + if (!ctx.ExternalProcessorSubscribed) + { + return false; + } + + // second message - edit & retry + if (ctx.MessageSentCount == 0 && ctx.MessageHandledCount == 1) + { + var failedMessagedId = await this.GetOnlyFailedUnresolvedMessageId(); + if (failedMessagedId == null) + { + return false; + } + + ctx.OriginalMessageFailureId = failedMessagedId; + ctx.MessageSentCount = 1; + + string editedMessage = JsonSerializer.Serialize(new EditResolutionMessage + { }); + + SingleResult failedMessage = + await this.TryGet($"/api/errors/{ctx.OriginalMessageFailureId}"); + + var editModel = new EditMessageModel + { + MessageBody = editedMessage, + MessageHeaders = failedMessage.Item.ProcessingAttempts.Last().Headers + }; + await this.Post($"/api/edit/{ctx.OriginalMessageFailureId}", editModel); + return false; + } + + // third message - retry + if (ctx.MessageSentCount == 1 && ctx.MessageHandledCount == 2) + { + var failedMessageIdAfterId = await this.GetOnlyFailedUnresolvedMessageId(); + if (failedMessageIdAfterId == null) + { + return false; + } + + ctx.EditedMessageFailureId = failedMessageIdAfterId; + ctx.MessageSentCount = 2; + + await this.Post($"/api/errors/{ctx.EditedMessageFailureId}/retry"); + return false; + } + + if (ctx.MessageHandledCount != 3) + { + return false; + } + + if (!ctx.MessageResolved || !ctx.EditAndRetryHandled || !ctx.MessageFailedResolved) + { + return false; + } + + return true; + }).Run(); + + Assert.Multiple(() => + { + Assert.That(context.ResolvedMessageId, Is.EqualTo(context.EditedMessageFailureId)); + Assert.That(context.EditedMessageEditOf, Is.EqualTo(context.OriginalMessageFailureId)); + Assert.That(context.MessageFailedFailedMessageIds.Count, Is.EqualTo(2)); + Assert.That(context.MessageFailedFailedMessageIds, Is.Unique); + Assert.That(context.MessageFailedFailedMessageIds, Has.Some.EqualTo(context.OriginalMessageFailureId)); + Assert.That(context.RetryFailedMessageId, Is.EqualTo(context.OriginalMessageFailureId)); + Assert.That(context.MessageFailedFailedMessageIds, Has.Some.EqualTo(context.EditedMessageFailureId)); + }); + } + + + public class EditMessageResolutionContext : ScenarioContext + { + public string OriginalMessageFailureId { get; set; } + public int MessageSentCount { get; set; } + public int MessageHandledCount { get; set; } + public string ResolvedMessageId { get; set; } + public string EditedMessageFailureId { get; set; } + public string EditedMessageEditOf { get; set; } + public bool ExternalProcessorSubscribed { get; set; } + public bool MessageResolved { get; set; } + public bool MessageFailedResolved { get; set; } + public string RetryFailedMessageId { get; set; } + public bool EditAndRetryHandled { get; set; } + public List MessageFailedFailedMessageIds { get; } = []; + } + + public class MessageReceiver : EndpointConfigurationBuilder + { + public MessageReceiver() => EndpointSetup(c => c.NoRetries()); + + + public class EditMessageResolutionHandler(EditMessageResolutionContext testContext) + : IHandleMessages, IHandleMessages, IHandleMessages, IHandleMessages + { + public Task Handle(EditResolutionMessage message, IMessageHandlerContext context) + { + // First run - supposed to fail + if (testContext.MessageSentCount == 0) + { + testContext.MessageHandledCount = 1; + throw new SimulatedException(); + } + + // Second run - edit retry - supposed to fail + if (testContext.MessageSentCount == 1) + { + testContext.EditedMessageEditOf = context.MessageHeaders["ServiceControl.EditOf"]; + testContext.MessageHandledCount = 2; + throw new SimulatedException(); + } + + // Last run - normal retry - supposed to succeed + testContext.MessageHandledCount = 3; + return Task.CompletedTask; + } + + public Task Handle(MessageFailureResolvedByRetry message, IMessageHandlerContext context) + { + testContext.ResolvedMessageId = message.FailedMessageId; + testContext.MessageResolved = true; + return Task.CompletedTask; + } + + public Task Handle(MessageFailed message, IMessageHandlerContext context) + { + testContext.MessageFailedFailedMessageIds.Add(message.FailedMessageId); + if (testContext.MessageFailedFailedMessageIds.Count == 2) + { + testContext.MessageFailedResolved = true; + } + return Task.CompletedTask; + } + + public Task Handle(MessageEditedAndRetried message, IMessageHandlerContext context) + { + testContext.RetryFailedMessageId = message.FailedMessageId; + testContext.EditAndRetryHandled = true; + return Task.CompletedTask; + } + } + } + + public class EditResolutionMessage : IMessage + { + } + } +} + diff --git a/src/ServiceControl.AcceptanceTests/Recoverability/ExternalIntegration/When_a_failed_msg_is_resolved_by_edit.cs b/src/ServiceControl.AcceptanceTests/Recoverability/ExternalIntegration/When_a_failed_msg_is_resolved_by_edit.cs new file mode 100644 index 0000000000..a34a0e3c50 --- /dev/null +++ b/src/ServiceControl.AcceptanceTests/Recoverability/ExternalIntegration/When_a_failed_msg_is_resolved_by_edit.cs @@ -0,0 +1,154 @@ +namespace ServiceControl.AcceptanceTests.Recoverability.ExternalIntegration +{ + using System.Linq; + using System.Threading.Tasks; + using AcceptanceTesting; + using AcceptanceTesting.EndpointTemplates; + using Contracts; + using NServiceBus; + using NServiceBus.AcceptanceTesting; + using NUnit.Framework; + using ServiceControl.MessageFailures; + using ServiceControl.MessageFailures.Api; + using JsonSerializer = System.Text.Json.JsonSerializer; + + class When_a_failed_msg_is_resolved_by_edit : AcceptanceTest + { + [Test] + public async Task Should_publish_notification() + { + CustomConfiguration = config => config.OnEndpointSubscribed((s, ctx) => + { + ctx.ExternalProcessorSubscribed = s.SubscriberReturnAddress.Contains(nameof(MessageReceiver)); + }); + + var context = await Define() + .WithEndpoint(b => b.When(async (bus, c) => + { + await bus.Subscribe(); + await bus.Subscribe(); + + if (c.HasNativePubSubSupport) + { + c.ExternalProcessorSubscribed = true; + } + }).When(c => c.SendLocal(new EditResolutionMessage())).DoNotFailOnErrorMessages()) + .Done(async ctx => + { + if (!ctx.ExternalProcessorSubscribed) + { + return false; + } + + if (!ctx.OriginalMessageHandled) + { + return false; + } + + if (!ctx.EditedMessage) + { + var failedEditedMessage = await this.GetOnlyFailedUnresolvedMessageId(); + if (failedEditedMessage == null) + { + return false; + } + + ctx.OriginalMessageFailureId = failedEditedMessage; + + ctx.EditedMessage = true; + string editedMessage = JsonSerializer.Serialize(new EditResolutionMessage + { + HasBeenEdited = true + }); + + SingleResult failedMessage = + await this.TryGet($"/api/errors/{ctx.OriginalMessageFailureId}"); + + var editModel = new EditMessageModel + { + MessageBody = editedMessage, + MessageHeaders = failedMessage.Item.ProcessingAttempts.Last().Headers + }; + await this.Post($"/api/edit/{ctx.OriginalMessageFailureId}", editModel); + return false; + } + + if (!ctx.EditedMessageHandled) + { + return false; + } + + if (!ctx.MessageFailedHandled || !ctx.MessageEdited) + { + return false; + } + + return true; + }).Run(); + + Assert.Multiple(() => + { + Assert.That(context.EditedMessageId, Is.EqualTo(context.OriginalMessageFailureId)); + Assert.That(context.EditedMessageEditOf, Is.EqualTo(context.OriginalMessageFailureId)); + Assert.That(context.FailedMessageId, Is.EqualTo(context.OriginalMessageFailureId)); + }); + } + + + class EditMessageResolutionContext : ScenarioContext + { + public bool OriginalMessageHandled { get; set; } + public bool EditedMessage { get; set; } + public string OriginalMessageFailureId { get; set; } + public bool EditedMessageHandled { get; set; } + public string EditedMessageId { get; set; } + public string EditedMessageEditOf { get; set; } + public bool ExternalProcessorSubscribed { get; set; } + public bool MessageEdited { get; set; } + public string FailedMessageId { get; set; } + public bool MessageFailedHandled { get; set; } + } + + class MessageReceiver : EndpointConfigurationBuilder + { + public MessageReceiver() => EndpointSetup(c => c.NoRetries()); + + class EditMessageResolutionHandler(EditMessageResolutionContext testContext) : IHandleMessages, + IHandleMessages, + IHandleMessages + { + public Task Handle(EditResolutionMessage message, IMessageHandlerContext context) + { + if (message.HasBeenEdited) + { + testContext.EditedMessageEditOf = context.MessageHeaders["ServiceControl.EditOf"]; + testContext.EditedMessageHandled = true; + return Task.CompletedTask; + } + + testContext.OriginalMessageHandled = true; + throw new SimulatedException(); + } + + public Task Handle(MessageEditedAndRetried message, IMessageHandlerContext context) + { + testContext.EditedMessageId = message.FailedMessageId; + testContext.MessageEdited = true; + return Task.CompletedTask; + } + + public Task Handle(MessageFailed message, IMessageHandlerContext context) + { + testContext.FailedMessageId = message.FailedMessageId; + testContext.MessageFailedHandled = true; + return Task.CompletedTask; + } + } + } + + class EditResolutionMessage : IMessage + { + public bool HasBeenEdited { get; init; } + } + } +} \ No newline at end of file diff --git a/src/ServiceControl.AcceptanceTests/Recoverability/ExternalIntegration/When_a_reedit_solves_a_failed_msg.cs b/src/ServiceControl.AcceptanceTests/Recoverability/ExternalIntegration/When_a_reedit_solves_a_failed_msg.cs new file mode 100644 index 0000000000..0738fcae18 --- /dev/null +++ b/src/ServiceControl.AcceptanceTests/Recoverability/ExternalIntegration/When_a_reedit_solves_a_failed_msg.cs @@ -0,0 +1,211 @@ +namespace ServiceControl.AcceptanceTests.Recoverability.ExternalIntegration +{ + using System.Collections.Generic; + using System.Linq; + using System.Threading.Tasks; + using AcceptanceTesting; + using AcceptanceTesting.EndpointTemplates; + using Contracts; + using NServiceBus; + using NServiceBus.AcceptanceTesting; + using NUnit.Framework; + using ServiceControl.MessageFailures; + using ServiceControl.MessageFailures.Api; + using JsonSerializer = System.Text.Json.JsonSerializer; + + class When_a_reedit_solves_a_failed_msg : AcceptanceTest + { + [Test] + public async Task Should_publish_notification() + { + CustomConfiguration = config => config.OnEndpointSubscribed((s, ctx) => + { + ctx.ExternalProcessorSubscribed = s.SubscriberReturnAddress.Contains(nameof(MessageReceiver)); + }); + + var context = await Define() + .WithEndpoint(b => b.When(async (bus, c) => + { + await bus.Subscribe(); + await bus.Subscribe(); + await bus.Subscribe(); + + if (c.HasNativePubSubSupport) + { + c.ExternalProcessorSubscribed = true; + } + }).When(c => c.SendLocal(new EditResolutionMessage() { MessageAttempt = 0 })) + .DoNotFailOnErrorMessages()) + .Done(async ctx => + { + if (!ctx.ExternalProcessorSubscribed) + { + return false; + } + + if (!ctx.OriginalMessageHandled) + { + return false; + } + + if (!ctx.FirstEdit) + { + var failedMessagedId = await this.GetOnlyFailedUnresolvedMessageId(); + if (failedMessagedId == null) + { + return false; + } + + ctx.OriginalMessageFailureId = failedMessagedId; + ctx.FirstEdit = true; + + string editedMessage = + JsonSerializer.Serialize(new EditResolutionMessage { MessageAttempt = 1 }); + + SingleResult failedMessage = + await this.TryGet($"/api/errors/{ctx.OriginalMessageFailureId}"); + + var editModel = new EditMessageModel + { + MessageBody = editedMessage, + MessageHeaders = failedMessage.Item.ProcessingAttempts.Last().Headers + }; + await this.Post($"/api/edit/{ctx.OriginalMessageFailureId}", editModel); + return false; + } + + if (!ctx.FirstEditHandled) + { + return false; + } + + if (!ctx.SecondEdit) + { + var failedMessagedId = await this.GetOnlyFailedUnresolvedMessageId(); + if (failedMessagedId == null || failedMessagedId == ctx.OriginalMessageFailureId) + { + return false; + } + + ctx.SecondMessageFailureId = failedMessagedId; + ctx.SecondEdit = true; + + string editedMessage = + JsonSerializer.Serialize(new EditResolutionMessage { MessageAttempt = 2 }); + + SingleResult failedMessage = + await this.TryGet($"/api/errors/{ctx.SecondMessageFailureId}"); + + var editModel = new EditMessageModel + { + MessageBody = editedMessage, + MessageHeaders = failedMessage.Item.ProcessingAttempts.Last().Headers + }; + await this.Post($"/api/edit/{ctx.SecondMessageFailureId}", editModel); + return false; + } + + if (!ctx.SecondEditHandled) + { + return false; + } + + if (!ctx.MessageResolved || !ctx.EditAndRetryHandled) + { + return false; + } + + return true; + }).Run(); + + Assert.Multiple(() => + { + Assert.That(context.FailedMessageFailedMessageIds.Count, Is.EqualTo(2)); + Assert.That(context.FailedMessageFailedMessageIds, Is.Unique); + Assert.That(context.FailedMessageFailedMessageIds, Has.Some.EqualTo(context.OriginalMessageFailureId)); + Assert.That(context.EditedMessageEditOf1, Is.EqualTo(context.OriginalMessageFailureId)); + Assert.That(context.FailedMessageFailedMessageIds, Has.Some.EqualTo(context.OriginalMessageFailureId)); + Assert.That(context.RetryFailedMessageIds.Count, Is.EqualTo(2)); + Assert.That(context.RetryFailedMessageIds, Is.Unique); + Assert.That(context.RetryFailedMessageIds, Has.Some.EqualTo(context.OriginalMessageFailureId)); + Assert.That(context.EditedMessageEditOf2, Is.EqualTo(context.SecondMessageFailureId)); + Assert.That(context.RetryFailedMessageIds, Has.Some.EqualTo(context.SecondMessageFailureId)); + }); + } + + + public class EditMessageResolutionContext : ScenarioContext + { + public bool OriginalMessageHandled { get; set; } + public string OriginalMessageFailureId { get; set; } + public bool SecondEditHandled { get; set; } + public bool MessageResolved { get; set; } + public bool FirstEditHandled { get; set; } + public bool FirstEdit { get; set; } + public bool SecondEdit { get; set; } + public string SecondMessageFailureId { get; set; } + public string EditedMessageEditOf2 { get; set; } + public string EditedMessageEditOf1 { get; set; } + public bool ExternalProcessorSubscribed { get; set; } + public bool EditAndRetryHandled { get; set; } + public List RetryFailedMessageIds { get; } = []; + public List FailedMessageFailedMessageIds { get; } = []; + } + + public class MessageReceiver : EndpointConfigurationBuilder + { + public MessageReceiver() => EndpointSetup(c => c.NoRetries()); + + + public class EditMessageResolutionHandler(EditMessageResolutionContext testContext) + : IHandleMessages, IHandleMessages, IHandleMessages + { + public Task Handle(EditResolutionMessage message, IMessageHandlerContext context) + { + switch (message.MessageAttempt) + { + case 0: + testContext.OriginalMessageHandled = true; + throw new SimulatedException(); + case 1: + testContext.EditedMessageEditOf1 = context.MessageHeaders["ServiceControl.EditOf"]; + testContext.FirstEditHandled = true; + throw new SimulatedException(); + case 2: + testContext.EditedMessageEditOf2 = context.MessageHeaders["ServiceControl.EditOf"]; + testContext.SecondEditHandled = true; + return Task.CompletedTask; + default: + return Task.CompletedTask; + } + } + + public Task Handle(MessageFailed message, IMessageHandlerContext context) + { + testContext.FailedMessageFailedMessageIds.Add(message.FailedMessageId); + if (testContext.FailedMessageFailedMessageIds.Count == 2) + { + testContext.MessageResolved = true; + } + return Task.CompletedTask; + } + + public Task Handle(MessageEditedAndRetried message, IMessageHandlerContext context) + { + + testContext.RetryFailedMessageIds.Add(message.FailedMessageId); + if (testContext.RetryFailedMessageIds.Count == 2) + { + testContext.EditAndRetryHandled = true; + } + return Task.CompletedTask; + } + } + } + + public class EditResolutionMessage : IMessage + { + public int MessageAttempt { get; init; } + } + } +} \ No newline at end of file diff --git a/src/ServiceControl.AcceptanceTests/Recoverability/When_edited_message_fails_to_process.cs b/src/ServiceControl.AcceptanceTests/Recoverability/When_edited_message_fails_to_process.cs index 5c4b297b14..d668ba2a6d 100644 --- a/src/ServiceControl.AcceptanceTests/Recoverability/When_edited_message_fails_to_process.cs +++ b/src/ServiceControl.AcceptanceTests/Recoverability/When_edited_message_fails_to_process.cs @@ -1,15 +1,15 @@ namespace ServiceControl.AcceptanceTests.Recoverability { + using System; + using System.Collections.Generic; using System.Linq; using System.Text.Json; using System.Threading.Tasks; using AcceptanceTesting; using AcceptanceTesting.EndpointTemplates; using AcceptanceTests; - using Infrastructure; using NServiceBus; using NServiceBus.AcceptanceTesting; - using NServiceBus.Settings; using NUnit.Framework; using ServiceControl.MessageFailures; using ServiceControl.MessageFailures.Api; @@ -19,30 +19,55 @@ class When_edited_message_fails_to_process : AcceptanceTest [Test] public async Task A_new_message_failure_is_created() { + CustomConfiguration = config => config.OnEndpointSubscribed((s, ctx) => + { + ctx.ExternalProcessorSubscribed = s.SubscriberReturnAddress.Contains(nameof(FailingEditedMessageReceiver)); + }); + var context = await Define() - .WithEndpoint(e => e - .When(c => c.SendLocal(new FailingMessage())) + .WithEndpoint(b => b.When(async (bus, c) => + { + await bus.Subscribe(); + + if (c.HasNativePubSubSupport) + { + c.ExternalProcessorSubscribed = true; + } + }).When(c => c.SendLocal(new FailingMessage())) .DoNotFailOnErrorMessages()) .Done(async ctx => { - if (ctx.OriginalMessageFailureId == null) + if (!ctx.ExternalProcessorSubscribed) + { + return false; + } + + if (!ctx.OriginalMessageHandled) { return false; } if (!ctx.EditedMessage) { - var failedMessage = await this.TryGet($"/api/errors/{ctx.OriginalMessageFailureId}"); - if (!failedMessage.HasResult) + var failedMessageId = await this.GetOnlyFailedUnresolvedMessageId(); + if (failedMessageId == null) { return false; } + ctx.OriginalMessageFailureId = failedMessageId; + ctx.EditedMessage = true; + var editedMessageInternalId = Guid.NewGuid().ToString(); + ctx.EditedMessageInternalId = editedMessageInternalId; var editedMessage = JsonSerializer.Serialize(new FailingMessage { - HasBeenEdited = true + HasBeenEdited = true, + MessageInternalId = editedMessageInternalId }); + + var failedMessage = await this.TryGet($"/api/errors/{ctx.OriginalMessageFailureId}"); + var editModel = new EditMessageModel { MessageBody = editedMessage, @@ -53,23 +78,32 @@ public async Task A_new_message_failure_is_created() return false; } - if (ctx.EditedMessageFailureId == null) + if (!ctx.EditedMessageHandled || !ctx.MessageFailedHandled) { return false; } - var failedEditedMessage = await this.TryGet($"/api/errors/{ctx.EditedMessageFailureId}"); - if (!failedEditedMessage.HasResult) + var failedMessageIdAfterEdit = await this.GetOnlyFailedUnresolvedMessageId(); + if (failedMessageIdAfterEdit == null) { return false; } + if (failedMessageIdAfterEdit == ctx.OriginalMessageFailureId) + { + return false; + } + + ctx.EditedMessageFailureId = failedMessageIdAfterEdit; + ctx.OriginalMessageFailure = (await this.TryGet($"/api/errors/{ctx.OriginalMessageFailureId}")).Item; ctx.EditedMessageFailure = (await this.TryGet($"/api/errors/{ctx.EditedMessageFailureId}")).Item; return true; }) .Run(); + var editedMessageBody = JsonSerializer.Deserialize(context.EditedMessageFailure.ProcessingAttempts.Last().MessageMetadata["MsgFullText"].ToString()); + Assert.Multiple(() => { Assert.That(context.OriginalMessageFailure.Id, Is.Not.EqualTo(context.EditedMessageFailure.Id)); @@ -79,44 +113,66 @@ public async Task A_new_message_failure_is_created() Assert.That( "FailedMessages/" + context.EditedMessageFailure.ProcessingAttempts.Last().Headers["ServiceControl.EditOf"], Is.EqualTo(context.OriginalMessageFailure.Id)); + Assert.That(editedMessageBody.MessageInternalId, Is.EqualTo(context.EditedMessageInternalId)); + Assert.That(context.MessageFailedIds, Has.Count.EqualTo(2)); + Assert.That(context.MessageFailedIds, Is.Unique); + Assert.That(context.MessageFailedIds, Has.Some.EqualTo(context.OriginalMessageFailureId)); + Assert.That(context.MessageFailedIds, Has.Some.EqualTo(context.EditedMessageFailureId)); }); } class EditMessageFailureContext : ScenarioContext { - public string OriginalMessageFailureId { get; set; } + public bool OriginalMessageHandled { get; set; } public bool EditedMessage { get; set; } - public string EditedMessageFailureId { get; set; } + public bool EditedMessageHandled { get; set; } public FailedMessage OriginalMessageFailure { get; set; } public FailedMessage EditedMessageFailure { get; set; } + public string OriginalMessageFailureId { get; set; } + public string EditedMessageFailureId { get; set; } + public string EditedMessageInternalId { get; set; } + public bool ExternalProcessorSubscribed { get; set; } + public bool MessageFailedHandled { get; set; } + public List MessageFailedIds { get; } = []; } class FailingEditedMessageReceiver : EndpointConfigurationBuilder { public FailingEditedMessageReceiver() => EndpointSetup(c => { c.NoRetries(); }); - class FailingMessageHandler(EditMessageFailureContext testContext, IReadOnlySettings settings) - : IHandleMessages + class FailingMessageHandler(EditMessageFailureContext testContext) + : IHandleMessages, IHandleMessages { public Task Handle(FailingMessage message, IMessageHandlerContext context) { if (message.HasBeenEdited) { - testContext.EditedMessageFailureId = DeterministicGuid.MakeId(context.MessageId, settings.EndpointName()).ToString(); + testContext.EditedMessageHandled = true; } else { - testContext.OriginalMessageFailureId = DeterministicGuid.MakeId(context.MessageId, settings.EndpointName()).ToString(); + testContext.OriginalMessageHandled = true; } throw new SimulatedException(); } + + public Task Handle(ServiceControl.Contracts.MessageFailed message, IMessageHandlerContext context) + { + testContext.MessageFailedIds.Add(message.FailedMessageId); + if (testContext.MessageFailedIds.Count == 2) + { + testContext.MessageFailedHandled = true; + } + return Task.CompletedTask; + } } } class FailingMessage : IMessage { - public bool HasBeenEdited { get; set; } + public bool HasBeenEdited { get; init; } + public string MessageInternalId { get; init; } } } } \ No newline at end of file diff --git a/src/ServiceControl.Persistence.RavenDB/Editing/EditFailedMessageManager.cs b/src/ServiceControl.Persistence.RavenDB/Editing/EditFailedMessageManager.cs index e6b9cfb3a6..eb1c722286 100644 --- a/src/ServiceControl.Persistence.RavenDB/Editing/EditFailedMessageManager.cs +++ b/src/ServiceControl.Persistence.RavenDB/Editing/EditFailedMessageManager.cs @@ -25,13 +25,13 @@ public async Task GetFailedMessage(string failedMessageId) return failedMessage; } - public async Task GetCurrentEditingMessageId(string failedMessageId) + public async Task GetCurrentEditingRequestId(string failedMessageId) { var edit = await session.LoadAsync(FailedMessageEdit.MakeDocumentId(failedMessageId)); return edit?.EditId; } - public Task SetCurrentEditingMessageId(string editingMessageId) + public Task SetCurrentEditingRequestId(string editingMessageId) { if (failedMessage == null) { diff --git a/src/ServiceControl.Persistence.Tests/Recoverability/EditMessageTests.cs b/src/ServiceControl.Persistence.Tests/Recoverability/EditMessageTests.cs index f28c43a214..38f73a2edc 100644 --- a/src/ServiceControl.Persistence.Tests/Recoverability/EditMessageTests.cs +++ b/src/ServiceControl.Persistence.Tests/Recoverability/EditMessageTests.cs @@ -58,7 +58,7 @@ public async Task Should_discard_edit_if_edited_message_not_unresolved(FailedMes var failedMessage = await ErrorMessageDataStore.ErrorBy(failedMessageId); var editFailedMessagesManager = await ErrorMessageDataStore.CreateEditFailedMessageManager(); - var editOperation = await editFailedMessagesManager.GetCurrentEditingMessageId(failedMessageId); + var editOperation = await editFailedMessagesManager.GetCurrentEditingRequestId(failedMessageId); Assert.Multiple(() => { @@ -79,7 +79,7 @@ public async Task Should_discard_edit_when_different_edit_already_exists() using (var editFailedMessagesManager = await ErrorMessageDataStore.CreateEditFailedMessageManager()) { _ = await editFailedMessagesManager.GetFailedMessage(failedMessageId); - await editFailedMessagesManager.SetCurrentEditingMessageId(previousEdit); + await editFailedMessagesManager.SetCurrentEditingRequestId(previousEdit); await editFailedMessagesManager.SaveChanges(); } @@ -91,7 +91,7 @@ public async Task Should_discard_edit_when_different_edit_already_exists() using (var editFailedMessagesManagerAssert = await ErrorMessageDataStore.CreateEditFailedMessageManager()) { var failedMessage = await editFailedMessagesManagerAssert.GetFailedMessage(failedMessageId); - var editId = await editFailedMessagesManagerAssert.GetCurrentEditingMessageId(failedMessageId); + var editId = await editFailedMessagesManagerAssert.GetCurrentEditingRequestId(failedMessageId); Assert.Multiple(() => { @@ -130,7 +130,7 @@ public async Task Should_dispatch_edited_message_when_first_edit() var failedMessage2 = await x.GetFailedMessage(failedMessage.UniqueMessageId); Assert.That(failedMessage2, Is.Not.Null, "Edited failed message"); - var editId = await x.GetCurrentEditingMessageId(failedMessage2.UniqueMessageId); + var editId = await x.GetCurrentEditingRequestId(failedMessage2.UniqueMessageId); Assert.Multiple(() => { diff --git a/src/ServiceControl.Persistence/IEditFailedMessagesManager.cs b/src/ServiceControl.Persistence/IEditFailedMessagesManager.cs index 4e58d8ed05..f5ab6fdbb2 100644 --- a/src/ServiceControl.Persistence/IEditFailedMessagesManager.cs +++ b/src/ServiceControl.Persistence/IEditFailedMessagesManager.cs @@ -6,8 +6,8 @@ public interface IEditFailedMessagesManager : IDataSessionManager { Task GetFailedMessage(string failedMessageId); - Task GetCurrentEditingMessageId(string failedMessageId); - Task SetCurrentEditingMessageId(string editingMessageId); + Task GetCurrentEditingRequestId(string failedMessageId); + Task SetCurrentEditingRequestId(string editingMessageId); Task SetFailedMessageAsResolved(); } } diff --git a/src/ServiceControl/Contracts/MessageFailures/MessageEditedAndRetried.cs b/src/ServiceControl/Contracts/MessageFailures/MessageEditedAndRetried.cs new file mode 100644 index 0000000000..4a64221e13 --- /dev/null +++ b/src/ServiceControl/Contracts/MessageFailures/MessageEditedAndRetried.cs @@ -0,0 +1,9 @@ +namespace ServiceControl.Contracts.MessageFailures +{ + using Infrastructure.DomainEvents; + + public class MessageEditedAndRetried : IDomainEvent + { + public string FailedMessageId { get; set; } + } +} \ No newline at end of file diff --git a/src/ServiceControl/MessageFailures/Api/EditFailedMessagesController.cs b/src/ServiceControl/MessageFailures/Api/EditFailedMessagesController.cs index 533f2c8792..3ce42fc3ff 100644 --- a/src/ServiceControl/MessageFailures/Api/EditFailedMessagesController.cs +++ b/src/ServiceControl/MessageFailures/Api/EditFailedMessagesController.cs @@ -37,7 +37,7 @@ public async Task> Edit(string failedMessageId, //HINT: This validation is the first one because we want to minimize the chance of two users concurrently execute an edit-retry. var editManager = await store.CreateEditFailedMessageManager(); - var editId = await editManager.GetCurrentEditingMessageId(failedMessageId); + var editId = await editManager.GetCurrentEditingRequestId(failedMessageId); if (editId != null) { logger.LogWarning("Cannot edit message {FailedMessageId} because it has already been edited", failedMessageId); diff --git a/src/ServiceControl/Recoverability/Editing/EditHandler.cs b/src/ServiceControl/Recoverability/Editing/EditHandler.cs index 86d01b7454..0390cd4844 100644 --- a/src/ServiceControl/Recoverability/Editing/EditHandler.cs +++ b/src/ServiceControl/Recoverability/Editing/EditHandler.cs @@ -3,6 +3,8 @@ using System; using System.Linq; using System.Threading.Tasks; + using Contracts.MessageFailures; + using Infrastructure.DomainEvents; using MessageFailures; using Microsoft.Extensions.Logging; using NServiceBus; @@ -14,12 +16,13 @@ class EditHandler : IHandleMessages { - public EditHandler(IErrorMessageDataStore store, IMessageRedirectsDataStore redirectsStore, IMessageDispatcher dispatcher, ErrorQueueNameCache errorQueueNameCache, ILogger logger) + public EditHandler(IErrorMessageDataStore store, IMessageRedirectsDataStore redirectsStore, IMessageDispatcher dispatcher, ErrorQueueNameCache errorQueueNameCache, IDomainEvents domainEvents, ILogger logger) { this.store = store; this.redirectsStore = redirectsStore; this.dispatcher = dispatcher; this.errorQueueNameCache = errorQueueNameCache; + this.domainEvents = domainEvents; this.logger = logger; corruptedReplyToHeaderStrategy = new CorruptedReplyToHeaderStrategy(RuntimeEnvironment.MachineName, logger); @@ -28,6 +31,7 @@ public EditHandler(IErrorMessageDataStore store, IMessageRedirectsDataStore redi public async Task Handle(EditAndSend message, IMessageHandlerContext context) { FailedMessage failedMessage; + string editId; using (var session = await store.CreateEditFailedMessageManager()) { failedMessage = await session.GetFailedMessage(message.FailedMessageId); @@ -38,7 +42,7 @@ public async Task Handle(EditAndSend message, IMessageHandlerContext context) return; } - var editId = await session.GetCurrentEditingMessageId(message.FailedMessageId); + editId = await session.GetCurrentEditingRequestId(message.FailedMessageId); if (editId == null) { if (failedMessage.Status != FailedMessageStatus.Unresolved) @@ -48,7 +52,7 @@ public async Task Handle(EditAndSend message, IMessageHandlerContext context) } // create a retries document to prevent concurrent edits - await session.SetCurrentEditingMessageId(context.MessageId); + await session.SetCurrentEditingRequestId(context.MessageId); } else if (editId != context.MessageId) { @@ -80,6 +84,11 @@ public async Task Handle(EditAndSend message, IMessageHandlerContext context) address = retryTo; } await DispatchEditedMessage(outgoingMessage, address, context); + + await domainEvents.Raise(new MessageEditedAndRetried + { + FailedMessageId = message.FailedMessageId + }, context.CancellationToken); } OutgoingMessage BuildMessage(EditAndSend message) @@ -122,5 +131,6 @@ Task DispatchEditedMessage(OutgoingMessage editedMessage, string address, IMessa readonly IMessageDispatcher dispatcher; readonly ErrorQueueNameCache errorQueueNameCache; readonly ILogger logger; + readonly IDomainEvents domainEvents; } } \ No newline at end of file diff --git a/src/ServiceControl/Recoverability/ExternalIntegration/MessageEditedAndRetriedPublisher.cs b/src/ServiceControl/Recoverability/ExternalIntegration/MessageEditedAndRetriedPublisher.cs new file mode 100644 index 0000000000..a737606e10 --- /dev/null +++ b/src/ServiceControl/Recoverability/ExternalIntegration/MessageEditedAndRetriedPublisher.cs @@ -0,0 +1,29 @@ +namespace ServiceControl.Recoverability.ExternalIntegration; + +using System.Collections.Generic; +using System.Linq; +using System.Threading.Tasks; +using Contracts.MessageFailures; +using ExternalIntegrations; + +class MessageEditedAndRetriedPublisher : EventPublisher +{ + protected override DispatchContext CreateDispatchRequest(MessageEditedAndRetried @event) => + new() + { + FailedMessageId = @event.FailedMessageId, + }; + + protected override Task> PublishEvents(IEnumerable contexts) => + Task.FromResult(contexts.Select(x => (object)new Contracts.MessageEditedAndRetried + { + FailedMessageId = x.FailedMessageId, + })); + + public class DispatchContext + { + public string FailedMessageId { get; set; } + } + +} \ No newline at end of file diff --git a/src/ServiceControl/Recoverability/RecoverabilityComponent.cs b/src/ServiceControl/Recoverability/RecoverabilityComponent.cs index 8c52e4577b..8459008bb4 100644 --- a/src/ServiceControl/Recoverability/RecoverabilityComponent.cs +++ b/src/ServiceControl/Recoverability/RecoverabilityComponent.cs @@ -87,6 +87,7 @@ public override void Configure(Settings settings, ITransportCustomization transp services.AddIntegrationEventPublisher(); services.AddIntegrationEventPublisher(); services.AddIntegrationEventPublisher(); + services.AddIntegrationEventPublisher(); //Event log services.AddEventLogMapping();