diff --git a/Test/DurableTask.AzureStorage.Tests/PoisonMessageHandlingTests.cs b/Test/DurableTask.AzureStorage.Tests/PoisonMessageHandlingTests.cs new file mode 100644 index 000000000..3c1a63246 --- /dev/null +++ b/Test/DurableTask.AzureStorage.Tests/PoisonMessageHandlingTests.cs @@ -0,0 +1,1019 @@ +// ---------------------------------------------------------------------------------- +// Copyright Microsoft Corporation +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// http://www.apache.org/licenses/LICENSE-2.0 +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ---------------------------------------------------------------------------------- + +namespace DurableTask.AzureStorage.Tests +{ + using System; + using System.Collections.Generic; + using System.Diagnostics; + using System.Runtime.Serialization; + using System.Threading; + using System.Threading.Tasks; + using Azure.Storage.Blobs; + using Azure.Storage.Blobs.Models; + using DurableTask.Core; + using DurableTask.Core.Entities; + using DurableTask.Core.History; + using Microsoft.VisualStudio.TestTools.UnitTesting; + using Newtonsoft.Json; + + /// + /// Integration tests for poison message handling in . + /// These tests require the Azure Storage emulator (Azurite) to be running. + /// + /// + /// Because is sealed, these tests use a + /// decorator to force a transient failure the first time a work + /// item is completed. That failure causes the underlying message to be abandoned and redelivered, which increments + /// its dequeue count (and therefore its ) so that it exceeds the configured + /// maximum dispatch count and is treated as a poison message. + /// + [TestClass] + public class PoisonMessageHandlingTests + { + static readonly TimeSpan DefaultTimeout = Debugger.IsAttached ? TimeSpan.FromMinutes(5) : TimeSpan.FromSeconds(60); + + [TestMethod] + public async Task OrchestrationWithPoisonMessage_Failed_AndPoisonMessageStored() + { + string prefix = CreateUniquePrefix(); + AzureStorageOrchestrationServiceSettings settings = CreateSettings(MaxDequeueCount: 1, prefix: prefix); + + BlobServiceClient blobServiceClient = CreateBlobServiceClient(); + string containerName = GetPoisonContainerName(settings); + BlobContainerClient containerClient = blobServiceClient.GetBlobContainerClient(containerName); + await containerClient.DeleteIfExistsAsync(); + + var inner = new AzureStorageOrchestrationService(settings); + + // Fail the first orchestration work item completion so the ExecutionStarted message is redelivered + // with a dequeue count of 2, which exceeds the maximum dequeue count of 1. + var service = new FaultInjectingOrchestrationService(inner) + { + OrchestrationCompletionFailuresRemaining = 1, + }; + + await service.CreateAsync(recreateInstanceStore: true); + + using var worker = new TaskHubWorker(service, loggerFactory: settings.LoggerFactory); + worker.ErrorPropagationMode = ErrorPropagationMode.UseFailureDetails; + worker.AddTaskOrchestrations(typeof(EchoOrchestration)); + await worker.StartAsync(); + + try + { + var client = new TaskHubClient(service, loggerFactory: settings.LoggerFactory); + var tags = new Dictionary { { "key", "value" } }; + OrchestrationInstance instance = await client.CreateOrchestrationInstanceAsync( + name: NameVersionHelper.GetDefaultName(typeof(EchoOrchestration)), + version: NameVersionHelper.GetDefaultVersion(typeof(EchoOrchestration)), + instanceId: Guid.NewGuid().ToString("N"), + input: "hello", + tags: tags); + + // The orchestration itself never completes because the ExecutionStartedEvent was discarded, so we wait + // for the poison blob to appear rather than for orchestration completion. + await TestHelpers.WaitFor( + () => containerClient.Exists().Value && ListBlobsAsync(containerClient).GetAwaiter().GetResult().Count > 0, + TimeSpan.FromSeconds(30)); + + List blobs = await ListBlobsAsync(containerClient); + Assert.AreEqual(1, blobs.Count); + + // Control-queue (orchestration) poison messages are stored under the "instance-messages/" prefix. + string expectedPrefix = $"{InstanceMessagesBlobPrefix}{instance.InstanceId}_{instance.ExecutionId}"; + Assert.AreEqual(expectedPrefix, blobs[0].Name.Substring(0, expectedPrefix.Length)); + + MessageData poisonMessage = await DownloadPoisonMessagesAsync(containerClient, blobs[0].Name, CreateMessageManager(settings)); + Assert.AreEqual(string.Empty, poisonMessage.Sender.InstanceId); + Assert.AreEqual(string.Empty, poisonMessage.Sender.ExecutionId); + Assert.AreEqual(instance.InstanceId, poisonMessage.TaskMessage.OrchestrationInstance.InstanceId); + Assert.AreEqual(instance.ExecutionId, poisonMessage.TaskMessage.OrchestrationInstance.ExecutionId); + + Assert.IsInstanceOfType(poisonMessage.TaskMessage.Event, typeof(ExecutionStartedEvent)); + var executionStartedEvent = (ExecutionStartedEvent)poisonMessage.TaskMessage.Event; + Assert.AreEqual(instance.InstanceId, executionStartedEvent.OrchestrationInstance.InstanceId); + Assert.AreEqual(instance.ExecutionId, executionStartedEvent.OrchestrationInstance.ExecutionId); + Assert.AreEqual(NameVersionHelper.GetDefaultName(typeof(EchoOrchestration)), executionStartedEvent.Name); + Assert.AreEqual("\"hello\"", executionStartedEvent.Input); + Assert.IsNotNull(executionStartedEvent.Tags); + Assert.AreEqual(1, executionStartedEvent.Tags.Count); + Assert.IsTrue(executionStartedEvent.Tags.ContainsKey("key")); + Assert.AreEqual("value", executionStartedEvent.Tags["key"]); + + await AssertQueuesAreEmptyAsync(settings); + } + finally + { + await worker.StopAsync(isForced: true); + await containerClient.DeleteIfExistsAsync(); + await service.DeleteAsync(); + } + } + + static IEnumerable InvalidInstanceIdCases() + { + // A single invalid control character (U+0080) is replaced with a dash. + yield return new object[] { "bad\u0080id", "bad-id" }; + + // Multiple invalid control characters (U+0080 and U+008E) are each replaced with a dash. + yield return new object[] { "a\u0080b\u008Ec", "a-b-c" }; + + // An instance ID that is too long for a blob name is truncated. The sanitized value is unchanged (all + // characters are valid) but the composed blob name prefix exceeds the length limit and must be cut. + yield return new object[] { new string('a', 1000), new string('a', 1000) }; + + // A blob name may contain at most 254 '/' path segment delimiters. The blob name already contains one + // '/' from the "instance-messages/" prefix, so only the first 253 '/' from the instance ID are kept; + // every '/' after that is replaced with a dash. + const int slashCount = 300; + yield return new object[] + { + new string('/', slashCount), + new string('/', 253) + new string('-', slashCount - 253), + }; + } + + [DataTestMethod] + [DynamicData(nameof(InvalidInstanceIdCases), DynamicDataSourceType.Method)] + public async Task OrchestrationWithPoisonMessage_InvalidInstanceId_IsSanitizedInBlobName( + string instanceId, + string expectedSanitizedInstanceId) + { + string prefix = CreateUniquePrefix(); + + // MaxDequeueCount of 0 causes the message to be treated as poison on its very first dequeue, so we can + // enqueue a control message directly and observe the poison behavior without running an orchestration. + AzureStorageOrchestrationServiceSettings settings = CreateSettings(MaxDequeueCount: 0, prefix: prefix); + + BlobServiceClient blobServiceClient = CreateBlobServiceClient(); + string containerName = GetPoisonContainerName(settings); + BlobContainerClient instanceContainerClient = blobServiceClient.GetBlobContainerClient(containerName); + await instanceContainerClient.DeleteIfExistsAsync(); + + var service = new AzureStorageOrchestrationService(settings); + await service.CreateAsync(recreateInstanceStore: true); + + // Build and enqueue an ExecutionStarted control message with the (potentially invalid) instance ID + // directly onto the single control queue. This intentionally bypasses the tracking table, which cannot + // store arbitrarily long or exotic instance IDs, so we can focus on the poison-blob naming behavior. + + // Note that ExecutionStartedEvents can be enqueued before a table entry is created (for example for a suborchestration), + // so this path is valid to exercise + string executionId = Guid.NewGuid().ToString("N"); + var orchestrationInstance = new OrchestrationInstance { InstanceId = instanceId, ExecutionId = executionId }; + var executionStartedEvent = new ExecutionStartedEvent(-1, "\"hello\"") + { + Name = "SomeOrchestration", + Version = string.Empty, + OrchestrationInstance = orchestrationInstance, + }; + var taskMessage = new TaskMessage + { + OrchestrationInstance = orchestrationInstance, + Event = executionStartedEvent, + }; + + string controlQueueName = AzureStorageOrchestrationService.GetControlQueueName(settings.TaskHubName, 0); + MessageManager messageManager = CreateMessageManager(settings); + var messageData = new MessageData( + taskMessage, + Guid.NewGuid(), + controlQueueName, + orchestrationEpisode: null, + sender: new OrchestrationInstance { InstanceId = string.Empty, ExecutionId = string.Empty }); + string body = await messageManager.SerializeMessageDataAsync(messageData); + + var azureStorageClient = new DurableTask.AzureStorage.Storage.AzureStorageClient(settings); + DurableTask.AzureStorage.Storage.Queue controlQueue = azureStorageClient.GetQueueReference(controlQueueName); + await controlQueue.AddMessageAsync(body, visibilityDelay: null); + + using var worker = new TaskHubWorker(service, loggerFactory: settings.LoggerFactory); + worker.AddTaskOrchestrations(typeof(EchoOrchestration)); + await worker.StartAsync(); + + try + { + await TestHelpers.WaitFor( + () => instanceContainerClient.Exists().Value && ListBlobsAsync(instanceContainerClient).GetAwaiter().GetResult().Count > 0, + TimeSpan.FromSeconds(30)); + + List blobs = await ListBlobsAsync(instanceContainerClient); + Assert.AreEqual(1, blobs.Count); + + // The blob name is "{sanitizedInstanceId}_{executionId}_{messageId}". The instance/execution portion + // is truncated so the total blob name stays within the 1024 character limit. Neither the sanitized + // instance IDs used here, the execution ID (hex GUID), nor the message ID contains an underscore, so + // the message ID is the final underscore-delimited segment. + string blobName = blobs[0].Name; + Assert.IsTrue(blobName.Length <= 1024, $"Blob name length {blobName.Length} exceeds the 1024 character limit."); + + int lastUnderscore = blobName.LastIndexOf('_'); + Assert.IsTrue(lastUnderscore > 0, "Blob name should contain a message ID segment."); + string actualPrefix = blobName.Substring(0, lastUnderscore); + string messageId = blobName.Substring(lastUnderscore + 1); + + string expectedPrefix = $"{InstanceMessagesBlobPrefix}{expectedSanitizedInstanceId}_{executionId}"; + int maxPrefixLength = 1024 - messageId.Length - 1; + if (expectedPrefix.Length > maxPrefixLength) + { + expectedPrefix = expectedPrefix.Substring(0, maxPrefixLength); + } + + Assert.AreEqual(expectedPrefix, actualPrefix); + + // The stored poison message is the raw queue message body. Sanitization only affects the blob name, + // so the stored content is exactly the original message body (which still contains the unsanitized + // instance ID). + string blobContent = await DownloadBlobTextAsync(instanceContainerClient, blobName); + Assert.AreEqual(body, blobContent); + + await AssertQueuesAreEmptyAsync(settings); + } + finally + { + await worker.StopAsync(isForced: true); + await instanceContainerClient.DeleteIfExistsAsync(); + await service.DeleteAsync(); + } + } + + [TestMethod] + public async Task OrchestrationWithMessageEqualToMaxDequeueCount_CompletesSuccessfully() + { + string prefix = CreateUniquePrefix(); + AzureStorageOrchestrationServiceSettings settings = CreateSettings(MaxDequeueCount: 2, prefix: prefix); + + BlobServiceClient blobServiceClient = CreateBlobServiceClient(); + string containerName = GetPoisonContainerName(settings); + BlobContainerClient containerClient = blobServiceClient.GetBlobContainerClient(containerName); + await containerClient.DeleteIfExistsAsync(); + + var inner = new AzureStorageOrchestrationService(settings); + + // Fail the first completion to force the dispatch count to 2, which is equal to (not greater than) the + // maximum dispatch count, so the message is not treated as poisoned. + var service = new FaultInjectingOrchestrationService(inner) + { + OrchestrationCompletionFailuresRemaining = 1, + }; + + await service.CreateAsync(recreateInstanceStore: true); + + using var worker = new TaskHubWorker(service, loggerFactory: settings.LoggerFactory); + worker.AddTaskOrchestrations(typeof(EchoOrchestration)); + await worker.StartAsync(); + + try + { + var client = new TaskHubClient(service, loggerFactory: settings.LoggerFactory); + OrchestrationInstance instance = await client.CreateOrchestrationInstanceAsync( + name: NameVersionHelper.GetDefaultName(typeof(EchoOrchestration)), + version: NameVersionHelper.GetDefaultVersion(typeof(EchoOrchestration)), + input: "hello"); + + OrchestrationState state = await client.WaitForOrchestrationAsync(instance, DefaultTimeout); + + Assert.IsNotNull(state); + Assert.AreEqual(OrchestrationStatus.Completed, state.OrchestrationStatus); + Assert.IsFalse(await containerClient.ExistsAsync(), $"Blob container '{containerName}' should not exist"); + } + finally + { + await worker.StopAsync(isForced: true); + await containerClient.DeleteIfExistsAsync(); + await service.DeleteAsync(); + } + } + + [TestMethod] + public async Task OrchestrationWithDequeueExceedingMax_PoisonHandlingDisabled_CompletesSuccessfully_NoBlob() + { + string prefix = CreateUniquePrefix(); + AzureStorageOrchestrationServiceSettings settings = CreateSettings(MaxDequeueCount: 1, prefix: prefix, poisonEnabled: false); + + BlobServiceClient blobServiceClient = CreateBlobServiceClient(); + string containerName = GetPoisonContainerName(settings); + BlobContainerClient containerClient = blobServiceClient.GetBlobContainerClient(containerName); + await containerClient.DeleteIfExistsAsync(); + + var inner = new AzureStorageOrchestrationService(settings); + var service = new FaultInjectingOrchestrationService(inner) + { + OrchestrationCompletionFailuresRemaining = 1, + }; + + await service.CreateAsync(recreateInstanceStore: true); + + using var worker = new TaskHubWorker(service, loggerFactory: settings.LoggerFactory); + worker.AddTaskOrchestrations(typeof(EchoOrchestration)); + await worker.StartAsync(); + + try + { + var client = new TaskHubClient(service, loggerFactory: settings.LoggerFactory); + OrchestrationInstance instance = await client.CreateOrchestrationInstanceAsync( + name: NameVersionHelper.GetDefaultName(typeof(EchoOrchestration)), + version: NameVersionHelper.GetDefaultVersion(typeof(EchoOrchestration)), + input: "hello"); + + OrchestrationState state = await client.WaitForOrchestrationAsync(instance, DefaultTimeout); + + Assert.IsNotNull(state); + Assert.AreEqual(OrchestrationStatus.Completed, state.OrchestrationStatus); + Assert.IsNull(state.FailureDetails); + + Assert.IsFalse( + await containerClient.ExistsAsync(), + $"Blob container '{containerName}' should not exist when poison handling is disabled"); + } + finally + { + await worker.StopAsync(isForced: true); + await service.DeleteAsync(); + } + } + + [TestMethod] + public async Task ActivityWithPoisonMessage_Failed_AndPoisonMessageStored() + { + string prefix = CreateUniquePrefix(); + AzureStorageOrchestrationServiceSettings settings = CreateSettings(MaxDequeueCount: 1, prefix: prefix); + + BlobServiceClient blobServiceClient = CreateBlobServiceClient(); + string containerName = GetPoisonContainerName(settings); + BlobContainerClient containerClient = blobServiceClient.GetBlobContainerClient(containerName); + await containerClient.DeleteIfExistsAsync(); + + var inner = new AzureStorageOrchestrationService(settings); + + // Fail the first activity work item completion so the TaskScheduled message is redelivered with a dispatch + // count of 2, which exceeds the maximum dispatch count of 1. + var service = new FaultInjectingOrchestrationService(inner) + { + ActivityCompletionFailuresRemaining = 1, + }; + + await service.CreateAsync(recreateInstanceStore: true); + + using var worker = new TaskHubWorker(service, loggerFactory: settings.LoggerFactory); + worker.ErrorPropagationMode = ErrorPropagationMode.UseFailureDetails; + worker.AddTaskOrchestrations(typeof(ScheduleActivityOrchestration)); + worker.AddTaskActivities(typeof(EchoActivity)); + await worker.StartAsync(); + + try + { + var client = new TaskHubClient(service, loggerFactory: settings.LoggerFactory); + OrchestrationInstance instance = await client.CreateOrchestrationInstanceAsync( + name: NameVersionHelper.GetDefaultName(typeof(ScheduleActivityOrchestration)), + version: NameVersionHelper.GetDefaultVersion(typeof(ScheduleActivityOrchestration)), + input: "hello"); + + // The activity message exceeds the maximum dequeue count and is moved to poison storage. The + // orchestration itself never completes because the activity result is never produced, so we wait + // for the poison blob to appear rather than for orchestration completion. + await TestHelpers.WaitFor( + () => containerClient.Exists().Value && ListBlobsAsync(containerClient).GetAwaiter().GetResult().Count > 0, + TimeSpan.FromSeconds(30)); + + List blobs = await ListBlobsAsync(containerClient); + Assert.AreEqual(1, blobs.Count); + + // Work item (activity) poison messages are stored under the "activity-messages/" prefix. + string expectedPrefix = $"{ActivityMessagesBlobPrefix}{instance.InstanceId}_{instance.ExecutionId}"; + Assert.AreEqual(expectedPrefix, blobs[0].Name.Substring(0, expectedPrefix.Length)); + + MessageData poisonMessage = await DownloadPoisonMessagesAsync(containerClient, blobs[0].Name, CreateMessageManager(settings)); + Assert.AreEqual(instance.InstanceId, poisonMessage.Sender.InstanceId); + Assert.AreEqual(instance.ExecutionId, poisonMessage.Sender.ExecutionId); + Assert.AreEqual(instance.InstanceId, poisonMessage.TaskMessage.OrchestrationInstance.InstanceId); + Assert.AreEqual(instance.ExecutionId, poisonMessage.TaskMessage.OrchestrationInstance.ExecutionId); + + string activityName = NameVersionHelper.GetDefaultName(typeof(EchoActivity)); + Assert.IsInstanceOfType(poisonMessage.TaskMessage.Event, typeof(TaskScheduledEvent)); + var taskScheduledEvent = (TaskScheduledEvent)poisonMessage.TaskMessage.Event; + Assert.AreEqual(activityName, taskScheduledEvent.Name); + Assert.AreEqual("[\"hello\"]", taskScheduledEvent.Input); + + await AssertQueuesAreEmptyAsync(settings); + } + finally + { + await worker.StopAsync(isForced: true); + await containerClient.DeleteIfExistsAsync(); + await service.DeleteAsync(); + } + } + + [TestMethod] + public async Task ActivityWithMessageEqualToMaxDequeueCount_CompletesSuccessfully() + { + string prefix = CreateUniquePrefix(); + AzureStorageOrchestrationServiceSettings settings = CreateSettings(MaxDequeueCount: 2, prefix: prefix); + + BlobServiceClient blobServiceClient = CreateBlobServiceClient(); + string activityContainerName = GetPoisonContainerName(settings); + BlobContainerClient activityContainerClient = blobServiceClient.GetBlobContainerClient(activityContainerName); + await activityContainerClient.DeleteIfExistsAsync(); + + var inner = new AzureStorageOrchestrationService(settings); + + // Fail the first activity completion to force the dispatch count to 2, which is equal to (not greater than) + // the maximum dispatch count, so the message is not treated as poisoned. + var service = new FaultInjectingOrchestrationService(inner) + { + ActivityCompletionFailuresRemaining = 1, + }; + + await service.CreateAsync(recreateInstanceStore: true); + + using var worker = new TaskHubWorker(service, loggerFactory: settings.LoggerFactory); + worker.AddTaskOrchestrations(typeof(ScheduleActivityOrchestration)); + worker.AddTaskActivities(typeof(EchoActivity)); + await worker.StartAsync(); + + try + { + var client = new TaskHubClient(service, loggerFactory: settings.LoggerFactory); + OrchestrationInstance instance = await client.CreateOrchestrationInstanceAsync( + name: NameVersionHelper.GetDefaultName(typeof(ScheduleActivityOrchestration)), + version: NameVersionHelper.GetDefaultVersion(typeof(ScheduleActivityOrchestration)), + input: "hello"); + + OrchestrationState state = await client.WaitForOrchestrationAsync(instance, DefaultTimeout); + + Assert.IsNotNull(state); + Assert.AreEqual(OrchestrationStatus.Completed, state.OrchestrationStatus); + Assert.IsFalse( + await activityContainerClient.ExistsAsync(), + $"Blob container '{activityContainerName}' should not exist"); + } + finally + { + await worker.StopAsync(isForced: true); + await activityContainerClient.DeleteIfExistsAsync(); + await service.DeleteAsync(); + } + } + + [TestMethod] + public async Task ActivityWithDispatchExceedingMax_PoisonHandlingDisabled_CompletesSuccessfully_NoBlob() + { + string prefix = CreateUniquePrefix(); + AzureStorageOrchestrationServiceSettings settings = CreateSettings(MaxDequeueCount: 1, prefix: prefix, poisonEnabled: false); + + BlobServiceClient blobServiceClient = CreateBlobServiceClient(); + string activityContainerName = GetPoisonContainerName(settings); + BlobContainerClient activityContainerClient = blobServiceClient.GetBlobContainerClient(activityContainerName); + await activityContainerClient.DeleteIfExistsAsync(); + + var inner = new AzureStorageOrchestrationService(settings); + var service = new FaultInjectingOrchestrationService(inner) + { + ActivityCompletionFailuresRemaining = 1, + }; + + await service.CreateAsync(recreateInstanceStore: true); + + using var worker = new TaskHubWorker(service, loggerFactory: settings.LoggerFactory); + worker.AddTaskOrchestrations(typeof(ScheduleActivityOrchestration)); + worker.AddTaskActivities(typeof(EchoActivity)); + await worker.StartAsync(); + + try + { + var client = new TaskHubClient(service, loggerFactory: settings.LoggerFactory); + OrchestrationInstance instance = await client.CreateOrchestrationInstanceAsync( + name: NameVersionHelper.GetDefaultName(typeof(ScheduleActivityOrchestration)), + version: NameVersionHelper.GetDefaultVersion(typeof(ScheduleActivityOrchestration)), + input: "hello"); + + OrchestrationState state = await client.WaitForOrchestrationAsync(instance, DefaultTimeout); + + Assert.IsNotNull(state); + Assert.AreEqual(OrchestrationStatus.Completed, state.OrchestrationStatus); + Assert.IsNull(state.FailureDetails); + + Assert.IsFalse( + await activityContainerClient.ExistsAsync(), + $"Blob container '{activityContainerName}' should not exist when poison handling is disabled"); + } + finally + { + await worker.StopAsync(isForced: true); + await service.DeleteAsync(); + } + } + + [TestMethod] + public async Task ControlQueueMessageWithBadJson_StoredAsPoison_AndDeleted() + { + string prefix = CreateUniquePrefix(); + AzureStorageOrchestrationServiceSettings settings = CreateSettings(MaxDequeueCount: 1, prefix: prefix); + + BlobServiceClient blobServiceClient = CreateBlobServiceClient(); + string containerName = GetPoisonContainerName(settings); + BlobContainerClient containerClient = blobServiceClient.GetBlobContainerClient(containerName); + await containerClient.DeleteIfExistsAsync(); + + var service = new AzureStorageOrchestrationService(settings); + await service.CreateAsync(recreateInstanceStore: true); + + // Insert a malformed message directly into the single control queue. + // its dequeue count exceeds the maximum it must be moved to poison storage and deleted from the queue. + var azureStorageClient = new DurableTask.AzureStorage.Storage.AzureStorageClient(settings); + string controlQueueName = AzureStorageOrchestrationService.GetControlQueueName(settings.TaskHubName, 0); + DurableTask.AzureStorage.Storage.Queue controlQueue = azureStorageClient.GetQueueReference(controlQueueName); + const string badMessage = "{ this is not valid json"; + await controlQueue.AddMessageAsync(badMessage, visibilityDelay: null); + + using var worker = new TaskHubWorker(service, loggerFactory: settings.LoggerFactory); + await worker.StartAsync(); + + try + { + await TestHelpers.WaitFor( + () => containerClient.Exists().Value && ListBlobsAsync(containerClient).GetAwaiter().GetResult().Count > 0, + TimeSpan.FromSeconds(30)); + + List blobs = await ListBlobsAsync(containerClient); + Assert.AreEqual(1, blobs.Count); + + // The malformed control-queue message is stored under the "instance-messages/" prefix. + Assert.IsTrue(blobs[0].Name.StartsWith(InstanceMessagesBlobPrefix)); + + // The stored poison message must match the original (undeserializable) queue message body. + string blobContent = await DownloadBlobTextAsync(containerClient, blobs[0].Name); + Assert.AreEqual(badMessage, blobContent); + + // The malformed message must have been deleted from the control queue. + await AssertQueuesAreEmptyAsync(settings); + } + finally + { + await worker.StopAsync(isForced: true); + await containerClient.DeleteIfExistsAsync(); + await service.DeleteAsync(); + } + } + + [TestMethod] + public async Task WorkItemQueueMessageWithBadJson_StoredAsPoison_AndDeleted() + { + string prefix = CreateUniquePrefix(); + AzureStorageOrchestrationServiceSettings settings = CreateSettings(MaxDequeueCount: 1, prefix: prefix); + + // The work item queue does not abandon undeserializable messages, so redelivery only happens after the + // visibility timeout expires. Shorten it so the message is quickly redelivered (bumping its dequeue + // count past the maximum) and moved to poison storage within the test's wait window. + settings.WorkItemQueueVisibilityTimeout = TimeSpan.FromSeconds(3); + + BlobServiceClient blobServiceClient = CreateBlobServiceClient(); + string containerName = GetPoisonContainerName(settings); + BlobContainerClient containerClient = blobServiceClient.GetBlobContainerClient(containerName); + await containerClient.DeleteIfExistsAsync(); + + var service = new AzureStorageOrchestrationService(settings); + await service.CreateAsync(recreateInstanceStore: true); + + // Insert a malformed message directly into the work item queue. + // dequeue count exceeds the maximum it must be moved to poison storage and deleted from the queue. + var azureStorageClient = new DurableTask.AzureStorage.Storage.AzureStorageClient(settings); + string workItemQueueName = AzureStorageOrchestrationService.GetWorkItemQueueName(settings.TaskHubName); + DurableTask.AzureStorage.Storage.Queue workItemQueue = azureStorageClient.GetQueueReference(workItemQueueName); + const string badMessage = "{ this is not valid json"; + await workItemQueue.AddMessageAsync(badMessage, visibilityDelay: null); + + using var worker = new TaskHubWorker(service, loggerFactory: settings.LoggerFactory); + await worker.StartAsync(); + + try + { + await TestHelpers.WaitFor( + () => containerClient.Exists().Value && ListBlobsAsync(containerClient).GetAwaiter().GetResult().Count > 0, + TimeSpan.FromSeconds(30)); + + List blobs = await ListBlobsAsync(containerClient); + Assert.AreEqual(1, blobs.Count); + + // The malformed work item message is stored under the "activity-messages/" prefix. + Assert.IsTrue(blobs[0].Name.StartsWith(ActivityMessagesBlobPrefix)); + + // The stored poison message must match the original (undeserializable) queue message body. + string blobContent = await DownloadBlobTextAsync(containerClient, blobs[0].Name); + Assert.AreEqual(badMessage, blobContent); + + // The malformed message must have been deleted from the work item queue. + await AssertQueuesAreEmptyAsync(settings); + } + finally + { + await worker.StopAsync(isForced: true); + await containerClient.DeleteIfExistsAsync(); + await service.DeleteAsync(); + } + } + + static AzureStorageOrchestrationServiceSettings CreateSettings( + int MaxDequeueCount, + string prefix, + bool poisonEnabled = true) + { + AzureStorageOrchestrationServiceSettings settings = TestHelpers.GetTestAzureStorageOrchestrationServiceSettings( + enableExtendedSessions: false); + + // Use a unique task hub per test to isolate queues/tables from other tests. + settings.TaskHubName = "poison" + Guid.NewGuid().ToString("N").Substring(0, 10); + settings.IsPoisonMessageStorageEnabled = poisonEnabled; + settings.MaxDequeueCount = MaxDequeueCount; + settings.PoisonMessageStorageContainerNameSuffix = prefix; + + // Use a single partition so tests have a single, deterministic control queue to inspect. + settings.PartitionCount = 1; + + return settings; + } + + // Verifies that every control queue and the work item queue for the given task hub is empty. This confirms + // that processed and poisoned messages have been removed from their source queues. + static async Task AssertQueuesAreEmptyAsync(AzureStorageOrchestrationServiceSettings settings) + { + var azureStorageClient = new DurableTask.AzureStorage.Storage.AzureStorageClient(settings); + + var queueNames = new List(); + for (int i = 0; i < settings.PartitionCount; i++) + { + queueNames.Add(AzureStorageOrchestrationService.GetControlQueueName(settings.TaskHubName, i)); + } + + queueNames.Add(AzureStorageOrchestrationService.GetWorkItemQueueName(settings.TaskHubName)); + + foreach (string queueName in queueNames) + { + DurableTask.AzureStorage.Storage.Queue queue = azureStorageClient.GetQueueReference(queueName); + await TestHelpers.WaitFor( + () => queue.GetApproximateMessagesCountAsync().GetAwaiter().GetResult() == 0, + TimeSpan.FromSeconds(30)); + } + } + + static BlobServiceClient CreateBlobServiceClient() + { + return new BlobServiceClient(TestHelpers.GetTestStorageAccountConnectionString()); + } + + // Poison messages for a task hub are stored in a single container named "{taskhubname}-{suffix}". + // Within that container, control-queue (orchestration/entity) messages are stored under the + // "instance-messages/" blob-name prefix and work item (activity) messages under "activity-messages/". + const string InstanceMessagesBlobPrefix = "instance-messages/"; + const string ActivityMessagesBlobPrefix = "activity-messages/"; + + static string GetPoisonContainerName(AzureStorageOrchestrationServiceSettings settings) => + $"{settings.TaskHubName.ToLowerInvariant()}-{settings.PoisonMessageStorageContainerNameSuffix}"; + + // Container names must be lowercase and no longer than 63 characters. The longest suffix we append is + // "-instance-messages" (18 characters), so keep the prefix short. + static string CreateUniquePrefix() + { + return "pmtest" + Guid.NewGuid().ToString("N").Substring(0, 12); + } + + static MessageManager CreateMessageManager(AzureStorageOrchestrationServiceSettings settings) + { + var azureStorageClient = new DurableTask.AzureStorage.Storage.AzureStorageClient(settings); + return new MessageManager(settings, azureStorageClient, "$root"); + } + + // When an orchestration fails, its FailureDetails may be surfaced directly on OrchestrationState.FailureDetails, + // or (in the current AzureStorage completion path) serialized into OrchestrationState.Output using the + // FailureDetails.ToString() format "{ErrorType}: {ErrorMessage}". This helper returns the FailureDetails from + // whichever the backend populated so tests can assert ErrorType/ErrorMessage uniformly. + static FailureDetails GetFailureDetails(OrchestrationState state) + { + if (state.FailureDetails != null) + { + return state.FailureDetails; + } + + string output = state.Output ?? string.Empty; + int separatorIndex = output.IndexOf(": ", StringComparison.Ordinal); + if (separatorIndex < 0) + { + return null; + } + + string errorType = output.Substring(0, separatorIndex); + string errorMessage = output.Substring(separatorIndex + 2); + + // The IsNonRetriable flag is not preserved in the flattened "{ErrorType}: {ErrorMessage}" output. The poison + // failure types are always written as non-retriable, so infer the flag from the error type. + bool isNonRetriable = errorType == "OrchestrationHistoryCorrupted" || errorType == "PoisonMessages"; + + return new FailureDetails(errorType, errorMessage, stackTrace: null, innerFailure: null, isNonRetriable: isNonRetriable); + } + + static async Task> ListBlobsAsync(BlobContainerClient containerClient) + { + var blobs = new List(); + await foreach (BlobItem blob in containerClient.GetBlobsAsync()) + { + blobs.Add(blob); + } + + return blobs; + } + + static async Task DownloadPoisonMessagesAsync(BlobContainerClient containerClient, string blobName, MessageManager messageManager) + { + BlobClient blobClient = containerClient.GetBlobClient(blobName); + BlobDownloadResult downloadResult = await blobClient.DownloadContentAsync(); + string blobContent = downloadResult.Content.ToString(); + + Assert.IsFalse(string.IsNullOrEmpty(blobContent), "Blob content should not be empty"); + + // The poison blob stores the raw queue message body, which is serialized with the product's message + // serializer (TypeNameHandling.Objects plus a custom type binder). Use a MessageManager so the same + // serializer settings/binder are applied when deserializing. + MessageData poisonMessage = messageManager.DeserializeMessageData(blobContent); + + Assert.IsNotNull(poisonMessage); + return poisonMessage; + } + + static async Task DownloadBlobTextAsync(BlobContainerClient containerClient, string blobName) + { + BlobClient blobClient = containerClient.GetBlobClient(blobName); + BlobDownloadResult downloadResult = await blobClient.DownloadContentAsync(); + return downloadResult.Content.ToString(); + } + + sealed class EchoOrchestration : TaskOrchestration + { + public override Task RunTask(OrchestrationContext context, string input) + { + return Task.FromResult(input); + } + } + + [KnownType(typeof(EchoActivity))] + sealed class ScheduleActivityOrchestration : TaskOrchestration + { + public override Task RunTask(OrchestrationContext context, string input) + { + return context.ScheduleTask(typeof(EchoActivity), input); + } + } + + sealed class EchoActivity : TaskActivity + { + protected override string Execute(TaskContext context, string input) + { + return input; + } + } + + [KnownType(typeof(ThrowingActivity))] + sealed class ScheduleThrowingActivityOrchestration : TaskOrchestration + { + public override Task RunTask(OrchestrationContext context, string input) + { + return context.ScheduleTask(typeof(ThrowingActivity), input); + } + } + + sealed class ThrowingActivity : TaskActivity + { + protected override string Execute(TaskContext context, string input) + { + throw new InvalidOperationException("boom from activity"); + } + } + + /// + /// A decorator around that forwards all calls to the inner + /// service but can be configured to throw a transient failure the first N times a work item is completed. + /// This is used to force work items to be abandoned and redelivered (incrementing their dequeue/dispatch + /// counts) so that poison message handling can be exercised. + /// + sealed class FaultInjectingOrchestrationService : IEntityOrchestrationService, IOrchestrationServiceClient + { + readonly AzureStorageOrchestrationService inner; + + int orchestrationCompletionFailuresRemaining; + int activityCompletionFailuresRemaining; + + public FaultInjectingOrchestrationService(AzureStorageOrchestrationService inner) + { + this.inner = inner ?? throw new ArgumentNullException(nameof(inner)); + } + + public int OrchestrationCompletionFailuresRemaining + { + get => this.orchestrationCompletionFailuresRemaining; + set => this.orchestrationCompletionFailuresRemaining = value; + } + + public int ActivityCompletionFailuresRemaining + { + get => this.activityCompletionFailuresRemaining; + set => this.activityCompletionFailuresRemaining = value; + } + + /// + /// Optional hook invoked on each activity work item after it is locked (but before it is dispatched), + /// allowing a test to corrupt the work item to exercise the invalid-work-item poison path. + /// + public Action CorruptActivityWorkItem { get; set; } + + /// + /// Optional hook invoked on each orchestration work item after it is locked (but before it is dispatched), + /// allowing a test to corrupt the work item to exercise the invalid-work-item poison path. + /// + public Action CorruptOrchestrationWorkItem { get; set; } + + IOrchestrationService InnerService => this.inner; + + IOrchestrationServiceClient InnerClient => this.inner; + + IEntityOrchestrationService InnerEntityService => this.inner; + + // ---- IOrchestrationService: management/lifecycle ---- + + public int TaskOrchestrationDispatcherCount => this.InnerService.TaskOrchestrationDispatcherCount; + + public int MaxConcurrentTaskOrchestrationWorkItems => this.InnerService.MaxConcurrentTaskOrchestrationWorkItems; + + public BehaviorOnContinueAsNew EventBehaviourForContinueAsNew => this.InnerService.EventBehaviourForContinueAsNew; + + public int TaskActivityDispatcherCount => this.InnerService.TaskActivityDispatcherCount; + + public int MaxConcurrentTaskActivityWorkItems => this.InnerService.MaxConcurrentTaskActivityWorkItems; + + public Task StartAsync() => this.InnerService.StartAsync(); + + public Task StopAsync() => this.InnerService.StopAsync(); + + public Task StopAsync(bool isForced) => this.InnerService.StopAsync(isForced); + + public Task CreateAsync() => this.InnerService.CreateAsync(); + + public Task CreateAsync(bool recreateInstanceStore) => this.InnerService.CreateAsync(recreateInstanceStore); + + public Task CreateIfNotExistsAsync() => this.InnerService.CreateIfNotExistsAsync(); + + public Task DeleteAsync() => this.InnerService.DeleteAsync(); + + public Task DeleteAsync(bool deleteInstanceStore) => this.InnerService.DeleteAsync(deleteInstanceStore); + + public bool IsMaxMessageCountExceeded(int currentMessageCount, OrchestrationRuntimeState runtimeState) => + this.InnerService.IsMaxMessageCountExceeded(currentMessageCount, runtimeState); + + public int GetDelayInSecondsAfterOnProcessException(Exception exception) => + this.InnerService.GetDelayInSecondsAfterOnProcessException(exception); + + public int GetDelayInSecondsAfterOnFetchException(Exception exception) => + this.InnerService.GetDelayInSecondsAfterOnFetchException(exception); + + // ---- IOrchestrationService: orchestration dispatcher ---- + + public async Task LockNextTaskOrchestrationWorkItemAsync(TimeSpan receiveTimeout, CancellationToken cancellationToken) + { + TaskOrchestrationWorkItem workItem = await this.InnerService.LockNextTaskOrchestrationWorkItemAsync(receiveTimeout, cancellationToken); + if (workItem != null) + { + this.CorruptOrchestrationWorkItem?.Invoke(workItem); + } + + return workItem; + } + + public Task RenewTaskOrchestrationWorkItemLockAsync(TaskOrchestrationWorkItem workItem) => + this.InnerService.RenewTaskOrchestrationWorkItemLockAsync(workItem); + + public Task CompleteTaskOrchestrationWorkItemAsync( + TaskOrchestrationWorkItem workItem, + OrchestrationRuntimeState newOrchestrationRuntimeState, + IList outboundMessages, + IList orchestratorMessages, + IList timerMessages, + TaskMessage continuedAsNewMessage, + OrchestrationState orchestrationState) + { + if (Interlocked.Decrement(ref this.orchestrationCompletionFailuresRemaining) >= 0) + { + throw new Exception("Simulated transient failure"); + } + + return this.InnerService.CompleteTaskOrchestrationWorkItemAsync( + workItem, + newOrchestrationRuntimeState, + outboundMessages, + orchestratorMessages, + timerMessages, + continuedAsNewMessage, + orchestrationState); + } + + public Task AbandonTaskOrchestrationWorkItemAsync(TaskOrchestrationWorkItem workItem) => + this.InnerService.AbandonTaskOrchestrationWorkItemAsync(workItem); + + public Task ReleaseTaskOrchestrationWorkItemAsync(TaskOrchestrationWorkItem workItem) => + this.InnerService.ReleaseTaskOrchestrationWorkItemAsync(workItem); + + // ---- IOrchestrationService: activity dispatcher ---- + + public Task LockNextTaskActivityWorkItem(TimeSpan receiveTimeout, CancellationToken cancellationToken) => + this.LockNextTaskActivityWorkItemInternalAsync(receiveTimeout, cancellationToken); + + async Task LockNextTaskActivityWorkItemInternalAsync(TimeSpan receiveTimeout, CancellationToken cancellationToken) + { + TaskActivityWorkItem workItem = await this.InnerService.LockNextTaskActivityWorkItem(receiveTimeout, cancellationToken); + if (workItem != null) + { + this.CorruptActivityWorkItem?.Invoke(workItem); + } + + return workItem; + } + + public Task RenewTaskActivityWorkItemLockAsync(TaskActivityWorkItem workItem) => + this.InnerService.RenewTaskActivityWorkItemLockAsync(workItem); + + public Task CompleteTaskActivityWorkItemAsync(TaskActivityWorkItem workItem, TaskMessage responseMessage) + { + if (Interlocked.Decrement(ref this.activityCompletionFailuresRemaining) >= 0) + { + throw new Exception("Simulated transient failure"); + } + + return this.InnerService.CompleteTaskActivityWorkItemAsync(workItem, responseMessage); + } + + public Task AbandonTaskActivityWorkItemAsync(TaskActivityWorkItem workItem) => + this.InnerService.AbandonTaskActivityWorkItemAsync(workItem); + + // ---- IEntityOrchestrationService ---- + + public EntityBackendProperties EntityBackendProperties => this.InnerEntityService.EntityBackendProperties; + + public EntityBackendQueries EntityBackendQueries => this.InnerEntityService.EntityBackendQueries; + + public async Task LockNextOrchestrationWorkItemAsync(TimeSpan receiveTimeout, CancellationToken cancellationToken) + { + TaskOrchestrationWorkItem workItem = await this.InnerEntityService.LockNextOrchestrationWorkItemAsync(receiveTimeout, cancellationToken); + if (workItem != null) + { + this.CorruptOrchestrationWorkItem?.Invoke(workItem); + } + + return workItem; + } + + public Task LockNextEntityWorkItemAsync(TimeSpan receiveTimeout, CancellationToken cancellationToken) => + this.InnerEntityService.LockNextEntityWorkItemAsync(receiveTimeout, cancellationToken); + + // ---- IOrchestrationServiceClient ---- + + public Task CreateTaskOrchestrationAsync(TaskMessage creationMessage) => + this.InnerClient.CreateTaskOrchestrationAsync(creationMessage); + + public Task CreateTaskOrchestrationAsync(TaskMessage creationMessage, OrchestrationStatus[] dedupeStatuses) => + this.InnerClient.CreateTaskOrchestrationAsync(creationMessage, dedupeStatuses); + + public Task SendTaskOrchestrationMessageAsync(TaskMessage message) => + this.InnerClient.SendTaskOrchestrationMessageAsync(message); + + public Task SendTaskOrchestrationMessageBatchAsync(params TaskMessage[] messages) => + this.InnerClient.SendTaskOrchestrationMessageBatchAsync(messages); + + public Task WaitForOrchestrationAsync(string instanceId, string executionId, TimeSpan timeout, CancellationToken cancellationToken) => + this.InnerClient.WaitForOrchestrationAsync(instanceId, executionId, timeout, cancellationToken); + + public Task ForceTerminateTaskOrchestrationAsync(string instanceId, string reason) => + this.InnerClient.ForceTerminateTaskOrchestrationAsync(instanceId, reason); + + public Task> GetOrchestrationStateAsync(string instanceId, bool allExecutions) => + this.InnerClient.GetOrchestrationStateAsync(instanceId, allExecutions); + + public Task GetOrchestrationStateAsync(string instanceId, string executionId) => + this.InnerClient.GetOrchestrationStateAsync(instanceId, executionId); + + public Task GetOrchestrationHistoryAsync(string instanceId, string executionId) => + this.InnerClient.GetOrchestrationHistoryAsync(instanceId, executionId); + + public Task PurgeOrchestrationHistoryAsync(DateTime thresholdDateTimeUtc, OrchestrationStateTimeRangeFilterType timeRangeFilterType) => + this.InnerClient.PurgeOrchestrationHistoryAsync(thresholdDateTimeUtc, timeRangeFilterType); + } + } +} diff --git a/src/DurableTask.AzureStorage/AnalyticsEventSource.cs b/src/DurableTask.AzureStorage/AnalyticsEventSource.cs index 38b6b2ba6..68bd57d4f 100644 --- a/src/DurableTask.AzureStorage/AnalyticsEventSource.cs +++ b/src/DurableTask.AzureStorage/AnalyticsEventSource.cs @@ -277,7 +277,7 @@ public void DuplicateMessageDetected( ExtensionVersion); } - [Event(EventIds.PoisonMessageDetected, Level = EventLevel.Warning, Version = 3)] + [Event(EventIds.PoisonMessageDetected, Level = EventLevel.Warning, Version = 4)] public void PoisonMessageDetected( string Account, string TaskHub, @@ -288,6 +288,7 @@ public void PoisonMessageDetected( string ExecutionId, string PartitionId, long DequeueCount, + string BlobName, string AppName, string ExtensionVersion) { @@ -302,6 +303,7 @@ public void PoisonMessageDetected( ExecutionId ?? string.Empty, PartitionId, DequeueCount, + BlobName, AppName, ExtensionVersion); } diff --git a/src/DurableTask.AzureStorage/AzureStorageOrchestrationServiceSettings.cs b/src/DurableTask.AzureStorage/AzureStorageOrchestrationServiceSettings.cs index e614da92f..a86c50f64 100644 --- a/src/DurableTask.AzureStorage/AzureStorageOrchestrationServiceSettings.cs +++ b/src/DurableTask.AzureStorage/AzureStorageOrchestrationServiceSettings.cs @@ -15,7 +15,6 @@ namespace DurableTask.AzureStorage { using System; using System.Runtime.Serialization; - using Azure.Data.Tables; using DurableTask.AzureStorage.Logging; using DurableTask.AzureStorage.Partitioning; using DurableTask.Core; @@ -311,5 +310,40 @@ internal LogHelper Logger /// table must now be read before processing every orchestration work item to obtain the latest etag. /// public bool UseInstanceTableEtag { get; set; } = false; + + /// + /// Gets or sets a flag whether poison message storage is enabled. + /// + /// + /// + /// If enabled, orchestration, entity, and activity messages that have been dequeued for processing + /// more than times will be moved to poison message storage in Azure Blob + /// Storage and deleted from their source queue. + /// This may leave orchestrations permanently as (or any other + /// non-terminal state) if the message(s) necessary for them make to progress are deemed "poisoned" and deleted. + /// + /// + public bool IsPoisonMessageStorageEnabled { get; set; } = true; + + /// + /// Gets or sets the number of times a message is dequeued for processing before it is considered "poisoned", + /// moved to poison message storage, and deleted from the source queue. The default value is 5,000, chosen with + /// the intent that only very pathological cases will be handled automatically. + /// + /// + /// This setting is applicable when is set to true. + /// + public long MaxDequeueCount { get; set; } = 5000; + + /// + /// Gets or sets the Azure Blob Storage container name suffix to use for poison message storage. + /// A container will be created with this suffix in the format "{taskhubname}-{suffix}". + /// The container name must adhere to the Azure Blob Storage container naming rules: + /// https://learn.microsoft.com/en-us/rest/api/storageservices/Naming-and-Referencing-Containers--Blobs--and-Metadata#directory-names + /// In particular, this means the total length of the name must not exceed 63 characters, the name can only contain lowercase letters, + /// numbers, and hyphens, must start and end with a letter or number, and cannot contain consecutive hyphens. + /// If not specified, the default value "poison-messages" will be used. + /// + public string PoisonMessageStorageContainerNameSuffix { get; set; } = "poison-messages"; } } diff --git a/src/DurableTask.AzureStorage/Logging/LogEvents.cs b/src/DurableTask.AzureStorage/Logging/LogEvents.cs index cd63f405a..ae7f3f572 100644 --- a/src/DurableTask.AzureStorage/Logging/LogEvents.cs +++ b/src/DurableTask.AzureStorage/Logging/LogEvents.cs @@ -666,7 +666,8 @@ public PoisonMessageDetected( string instanceId, string executionId, string partitionId, - long dequeueCount) + long dequeueCount, + string blobName) { this.Account = account; this.TaskHub = taskHub; @@ -677,6 +678,7 @@ public PoisonMessageDetected( this.ExecutionId = executionId; this.PartitionId = partitionId; this.DequeueCount = dequeueCount; + this.BlobName = blobName; } [StructuredLogField] @@ -706,18 +708,27 @@ public PoisonMessageDetected( [StructuredLogField] public long DequeueCount { get; } + [StructuredLogField] + public string BlobName { get; } + public override EventId EventId => new EventId( EventIds.PoisonMessageDetected, nameof(EventIds.PoisonMessageDetected)); public override LogLevel Level => LogLevel.Warning; - protected override string CreateLogMessage() => string.Format( - "{0}: Message {1} with ID {2} has been dequeued {3} times and is now considered poison", - this.InstanceId, - GetEventDescription(this.EventType, this.TaskEventId), - this.MessageId, - this.DequeueCount); + protected override string CreateLogMessage() + { + string message = string.Format( + "{0}: Message {1} with ID {2} has been dequeued {3} times and is now considered poison, stored in blob {4}", + this.InstanceId, + GetEventDescription(this.EventType, this.TaskEventId), + this.MessageId, + this.DequeueCount, + this.BlobName); + + return message; + } void IEventSourceEvent.WriteEventSource() => AnalyticsEventSource.Log.PoisonMessageDetected( this.Account, @@ -729,6 +740,7 @@ void IEventSourceEvent.WriteEventSource() => AnalyticsEventSource.Log.PoisonMess this.ExecutionId, this.PartitionId, this.DequeueCount, + this.BlobName, Utils.AppName, Utils.ExtensionVersion); } diff --git a/src/DurableTask.AzureStorage/Logging/LogHelper.cs b/src/DurableTask.AzureStorage/Logging/LogHelper.cs index e4ecaae14..6563df79f 100644 --- a/src/DurableTask.AzureStorage/Logging/LogHelper.cs +++ b/src/DurableTask.AzureStorage/Logging/LogHelper.cs @@ -236,7 +236,8 @@ internal void PoisonMessageDetected( string instanceId, string executionId, string partitionId, - long dequeueCount) + long dequeueCount, + string blobName) { var logEvent = new LogEvents.PoisonMessageDetected( account, @@ -247,7 +248,8 @@ internal void PoisonMessageDetected( instanceId, executionId, partitionId, - dequeueCount); + dequeueCount, + blobName); this.WriteStructuredLog(logEvent); } diff --git a/src/DurableTask.AzureStorage/Messaging/ControlQueue.cs b/src/DurableTask.AzureStorage/Messaging/ControlQueue.cs index 9f1c0d2ba..904f4b977 100644 --- a/src/DurableTask.AzureStorage/Messaging/ControlQueue.cs +++ b/src/DurableTask.AzureStorage/Messaging/ControlQueue.cs @@ -125,12 +125,31 @@ await batch.ParallelForEachAsync(async delegate (QueueMessage queueMessage) 0 /* TaskEventId */, e.ToString()); + if (await this.CheckForAndHandlePoisonMessageAsync( + blobNamePrefix: "instance-messages/", + queueMessage, + linkedCts.Token)) + { + return; + } + // Abandon the message so we can try it again later. // Note: We will fetch the message again from the queue before retrying, so no need to read the receipt _ = await this.AbandonMessageAsync(queueMessage); return; } + if (await this.CheckForAndHandlePoisonMessageAsync( + blobNamePrefix: "instance-messages/", + queueMessage, + linkedCts.Token, + messageData.TaskMessage.OrchestrationInstance, + messageData.TaskMessage.Event.EventType.ToString(), + Utils.GetTaskEventId(messageData.TaskMessage.Event))) + { + return; + } + // Check to see whether we've already dequeued this message. if (!this.stats.PendingOrchestratorMessages.TryAdd(queueMessage.MessageId, 1)) { diff --git a/src/DurableTask.AzureStorage/Messaging/TaskHubQueue.cs b/src/DurableTask.AzureStorage/Messaging/TaskHubQueue.cs index 221789566..693a9bfee 100644 --- a/src/DurableTask.AzureStorage/Messaging/TaskHubQueue.cs +++ b/src/DurableTask.AzureStorage/Messaging/TaskHubQueue.cs @@ -34,6 +34,10 @@ abstract class TaskHubQueue protected readonly string storageAccountName; protected readonly AzureStorageOrchestrationServiceSettings settings; protected readonly BackoffPollingHelper backoffHelper; + protected readonly string poisonMessageContainerName; + + BlobContainer? poisonMessageContainer; + public TaskHubQueue( AzureStorageClient azureStorageClient, @@ -55,6 +59,7 @@ public TaskHubQueue( } this.backoffHelper = new BackoffPollingHelper(minPollingDelay, maxPollingDelay); + this.poisonMessageContainerName = $"{this.settings.TaskHubName.ToLowerInvariant()}-{this.settings.PoisonMessageStorageContainerNameSuffix}"; } public string Name => this.storageQueue.Name; @@ -247,24 +252,10 @@ public virtual async Task AbandonMessageAsync(MessageData message, SessionBase? int taskEventId = taskMessage != null ? Utils.GetTaskEventId(taskMessage.Event) : -1; // Exponentially backoff a given queue message until a maximum visibility delay of 10 minutes. - // Once it hits the maximum, log the message as a poison message. const int maxSecondsToWait = 600; int numSecondsToWait = queueMessage.DequeueCount <= 30 ? Math.Min((int)Math.Pow(2, queueMessage.DequeueCount), maxSecondsToWait) : maxSecondsToWait; - if (numSecondsToWait == maxSecondsToWait) - { - this.settings.Logger.PoisonMessageDetected( - this.storageAccountName, - this.settings.TaskHubName, - eventType, - taskEventId, - queueMessage.MessageId, - instanceId, - executionId, - this.storageQueue.Name, - queueMessage.DequeueCount); - } this.settings.Logger.AbandoningMessage( this.storageAccountName, @@ -389,6 +380,184 @@ public virtual async Task DeleteMessageAsync(MessageData message, SessionBase? s } } + protected async Task CheckForAndHandlePoisonMessageAsync( + string blobNamePrefix, + QueueMessage queueMessage, + CancellationToken cancellationToken, + OrchestrationInstance? orchestrationInstance = null, + string eventType = "", + int taskEventId = -1) + { + if (!this.settings.IsPoisonMessageStorageEnabled || queueMessage.DequeueCount <= this.settings.MaxDequeueCount) + { + return false; + } + + try + { + BlobContainer container = this.poisonMessageContainer ??= + this.azureStorageClient.GetBlobContainerReference(this.poisonMessageContainerName); + + string blobName = CreateBlobName(orchestrationInstance, blobNamePrefix, queueMessage.MessageId); + + await container.CreateIfNotExistsAsync(cancellationToken); + + Blob blob = container.GetBlobReference(blobName); + await blob.UploadTextAsync(queueMessage.Body.ToString(), cancellationToken: cancellationToken); + + this.settings.Logger.PoisonMessageDetected( + this.storageAccountName, + this.settings.TaskHubName, + eventType, + taskEventId, + queueMessage.MessageId, + orchestrationInstance?.InstanceId ?? string.Empty, + orchestrationInstance?.ExecutionId ?? string.Empty, + this.storageQueue.Name, + queueMessage.DequeueCount, + blobName); + + await this.storageQueue.DeleteMessageAsync(queueMessage, cancellationToken: cancellationToken); + } + catch (Exception e) + { + this.settings.Logger.MessageFailure( + this.storageAccountName, + this.settings.TaskHubName, + queueMessage.MessageId, + orchestrationInstance?.InstanceId ?? string.Empty, + orchestrationInstance?.ExecutionId ?? string.Empty, + this.storageQueue.Name, + eventType, + taskEventId, + $"Error when attempting to store poison message. Error: {e}"); + + return false; + } + return true; + } + + static string CreateBlobName(OrchestrationInstance? orchestrationInstance, string blobNamePrefix, string messageId) + { + // Replace any invalid characters with a dash + string sanitizedInstanceId = SanitizeString(orchestrationInstance?.InstanceId, '-'); + blobNamePrefix += sanitizedInstanceId; + + if (!string.IsNullOrEmpty(orchestrationInstance?.ExecutionId)) + { + blobNamePrefix += $"_{SanitizeString(orchestrationInstance!.ExecutionId, '-')}"; + } + + // Blob name length limit is 1024 characters and we attach an extra character (_) and the message ID at the end + // From https://learn.microsoft.com/en-us/rest/api/storageservices/naming-and-referencing-containers--blobs--and-metadata?#blob-names + int maxPrefixLength = 1024 - messageId.Length - 1; + if (blobNamePrefix.Length > maxPrefixLength) + { + blobNamePrefix = blobNamePrefix.Substring(0, maxPrefixLength); + + // Avoid leaving an unpaired surrogate at the end of the truncated prefix. + if (char.IsHighSurrogate(blobNamePrefix, blobNamePrefix.Length - 1)) + { + blobNamePrefix = blobNamePrefix.Substring(0, blobNamePrefix.Length - 1); + } + } + string blobName = $"{blobNamePrefix}_{messageId}"; + + // A blob name may contain at most 254 path segment delimiters ('/'). Replace any '/' beyond the + // first 254 with a dash so the blob name remains valid. + // From https://learn.microsoft.com/en-us/rest/api/storageservices/naming-and-referencing-containers--blobs--and-metadata?#blob-names + const int MaxForwardSlashes = 254; + int forwardSlashCount = 0; + char[] blobNameChars = blobName.ToCharArray(); + for (int i = 0; i < blobNameChars.Length; i++) + { + if (blobNameChars[i] == '/' && ++forwardSlashCount > MaxForwardSlashes) + { + blobNameChars[i] = '-'; + } + } + return new string(blobNameChars); + } + + static string SanitizeString(string? input, char replacement) + { + if (input == null) + { + return string.Empty; + } + + // From https://learn.microsoft.com/en-us/rest/api/storageservices/naming-and-referencing-containers--blobs--and-metadata?#unicode-characters-not-recommended-for-use-in-container-or-blob-names + static bool IsInvalidCodePoint(int scalar) + { + if (scalar == 0x0080 || + (scalar >= 0x0082 && scalar <= 0x008C) || + scalar == 0x008E || + (scalar >= 0x0091 && scalar <= 0x009C) || + (scalar >= 0x009E && scalar <= 0x009F) || + (scalar >= 0xFDD1 && scalar <= 0xFDDC) || + (scalar >= 0xFDDE && scalar <= 0xFDEF) || + (scalar >= 0xFFF0 && scalar <= 0xFFFF)) + { + return true; + } + + return scalar == 0x1FFFE || scalar == 0x1FFFF || + scalar == 0x2FFFE || scalar == 0x2FFFF || + scalar == 0x3FFFE || scalar == 0x3FFFF || + scalar == 0x5FFFE || scalar == 0x5FFFF || + scalar == 0x6FFFE || scalar == 0x6FFFF || + scalar == 0x7FFFE || scalar == 0x7FFFF || + scalar == 0x9FFFE || scalar == 0x9FFFF || + scalar == 0xAFFFE || scalar == 0xAFFFF || + scalar == 0xBFFFE || scalar == 0xBFFFF || + scalar == 0xDFFFE || scalar == 0xDFFFF || + scalar == 0xEFFFE || scalar == 0xEFFFF || + scalar == 0xFFFFE || scalar == 0xFFFFF; + } + + var sb = new StringBuilder(input.Length); + + for (int i = 0; i < input.Length; i++) + { + char c = input[i]; + int scalar; + + // Since the .NET version is too low to iterate the runes, we manually detect and combine surrogate pairs here + if (char.IsHighSurrogate(c) && i + 1 < input.Length && char.IsLowSurrogate(input[i + 1])) + { + scalar = char.ConvertToUtf32(c, input[i + 1]); + i++; + } + else if (char.IsSurrogate(c)) + { + // Unpaired surrogate. These are not valid Unicode scalar values, so char.ConvertFromUtf32 below + // would throw for them. In practice an unpaired surrogate never reaches this method: instance and + // execution IDs are round-tripped through Azure Storage queues, which substitute unpaired + // surrogates (and other characters that are not valid in the storage encoding) with the Unicode + // replacement character (U+FFFD) before we read the message back. We therefore don't enumerate + // surrogate code points in IsInvalidCodePoint above; we just replace any that somehow appear here + // so that blob-name generation can never throw. + sb.Append(replacement); + continue; + } + else + { + scalar = c; + } + + if (IsInvalidCodePoint(scalar)) + { + sb.Append(replacement); + } + else + { + sb.Append(char.ConvertFromUtf32(scalar)); + } + } + + return sb.ToString(); + } + static bool IsMessageGoneException(Exception e) { DurableTaskStorageException? storageException = e as DurableTaskStorageException; diff --git a/src/DurableTask.AzureStorage/Messaging/WorkItemQueue.cs b/src/DurableTask.AzureStorage/Messaging/WorkItemQueue.cs index 512f761f5..d9fbf76f4 100644 --- a/src/DurableTask.AzureStorage/Messaging/WorkItemQueue.cs +++ b/src/DurableTask.AzureStorage/Messaging/WorkItemQueue.cs @@ -45,9 +45,39 @@ public async Task GetMessageAsync(CancellationToken cancellationTok continue; } - MessageData data = await this.messageManager.DeserializeQueueMessageAsync( + MessageData data; + + try + { + data = await this.messageManager.DeserializeQueueMessageAsync( + queueMessage, + this.storageQueue.Name); + } + catch (Exception) + { + if (await this.CheckForAndHandlePoisonMessageAsync( + blobNamePrefix: "activity-messages/", + queueMessage, + cancellationToken)) + { + this.backoffHelper.Reset(); + continue; + } + + throw; + } + + if (await this.CheckForAndHandlePoisonMessageAsync( + blobNamePrefix: "activity-messages/", queueMessage, - this.storageQueue.Name); + cancellationToken, + data.TaskMessage.OrchestrationInstance, + data.TaskMessage.Event.EventType.ToString(), + Utils.GetTaskEventId(data.TaskMessage.Event))) + { + this.backoffHelper.Reset(); + continue; + } this.backoffHelper.Reset(); return data; diff --git a/src/DurableTask.Core/History/ExecutionRewoundEvent.cs b/src/DurableTask.Core/History/ExecutionRewoundEvent.cs index c838e9eb2..4a9338156 100644 --- a/src/DurableTask.Core/History/ExecutionRewoundEvent.cs +++ b/src/DurableTask.Core/History/ExecutionRewoundEvent.cs @@ -39,6 +39,12 @@ public ExecutionRewoundEvent(int eventId, string? reason) this.Reason = reason; } + // Private ctor for JSON deserialization (required by some storage providers and out-of-proc executors) + ExecutionRewoundEvent() + : base(-1) + { + } + /// /// Gets the event type /// diff --git a/src/DurableTask.Core/Tracing/TraceHelper.cs b/src/DurableTask.Core/Tracing/TraceHelper.cs index 532e67a56..b139e83ac 100644 --- a/src/DurableTask.Core/Tracing/TraceHelper.cs +++ b/src/DurableTask.Core/Tracing/TraceHelper.cs @@ -626,51 +626,51 @@ internal static void EmitTraceActivityForTimer( /// the entity did not return results for all of the requests internal static void EndActivitiesForProcessingEntityInvocation(List traceActivities, List results, FailureDetails? batchFailureDetails) { - if (results.Count == traceActivities.Count) + foreach (var (activity, result) in traceActivities.Zip(results, (activity, result) => (activity, result))) { - foreach (var (activity, result) in traceActivities.Zip(results, (activity, result) => (activity, result))) + if (activity != null) { - if (activity != null) + if (result.ErrorMessage != null || result.FailureDetails != null) { - if (result.ErrorMessage != null || result.FailureDetails != null) - { - string errorDetails = result.ErrorMessage ?? result.FailureDetails!.ErrorMessage; - activity.SetTag(Schema.Task.ErrorMessage, errorDetails); - activity.SetStatus(ActivityStatusCode.Error, errorDetails); - } - else - { - activity.SetStatus(ActivityStatusCode.OK, "Completed"); - } - if (result.StartTimeUtc is DateTime startTime) - { - activity.SetStartTime(startTime); - } - if (result.EndTimeUtc is DateTime endTime) - { - activity.SetEndTime(endTime); - } - activity.Dispose(); + string errorDetails = result.ErrorMessage ?? result.FailureDetails!.ErrorMessage; + activity.SetTag(Schema.Task.ErrorMessage, errorDetails); + activity.SetStatus(ActivityStatusCode.Error, errorDetails); } + else + { + activity.SetStatus(ActivityStatusCode.OK, "Completed"); + } + if (result.StartTimeUtc is DateTime startTime) + { + activity.SetStartTime(startTime); + } + if (result.EndTimeUtc is DateTime endTime) + { + activity.SetEndTime(endTime); + } + activity.Dispose(); } } - // This can happen if some of the operations failed and have no corresponding OperationResult - // There is no way to map the successful operation results to the corresponding operation requests or trace activities, so we will just "fail" the trace activities in this case and dispose them - else + + // This can happen if not all of the operations in the batch were executed, in which case we populate the remaining + // activities with the failure details if they are available. + // If not, this work will be deferred and tried again, so we do not want to publish the activity. + for (int i = results.Count; i < traceActivities.Count; i++) { - string errorMessage = "Unable to generate a trace activity for the entity invocation even though it may have succeeded."; - if (batchFailureDetails is FailureDetails failureDetails) - { - errorMessage += $" If it failed, it may be due to {failureDetails.ErrorMessage}"; - } - foreach (var activity in traceActivities) + var activity = traceActivities[i]; + if (activity != null) { - if (activity != null) + if (batchFailureDetails != null) + { + activity.SetTag(Schema.Task.ErrorMessage, batchFailureDetails.ErrorMessage); + activity.SetStatus(ActivityStatusCode.Error, batchFailureDetails.ErrorMessage); + } + else { - activity.SetTag(Schema.Task.ErrorMessage, errorMessage); - activity.SetStatus(ActivityStatusCode.Error, errorMessage); - activity.Dispose(); + // This will only work if the listener honors the Recorded flag, so this is a best effort attempt + activity.ActivityTraceFlags &= ~ActivityTraceFlags.Recorded; } + activity.Dispose(); } } } @@ -943,4 +943,4 @@ static void ExceptionHandlingWrapper(Action innerFunc) } } } -} +} \ No newline at end of file