diff --git a/Test/DurableTask.AzureStorage.Tests/PoisonMessageHandlingTests.cs b/Test/DurableTask.AzureStorage.Tests/PoisonMessageHandlingTests.cs new file mode 100644 index 000000000..7f90224a6 --- /dev/null +++ b/Test/DurableTask.AzureStorage.Tests/PoisonMessageHandlingTests.cs @@ -0,0 +1,1608 @@ +// ---------------------------------------------------------------------------------- +// 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.ClientModel; + using System.Collections.Generic; + using System.Diagnostics; + using System.Linq; + using System.Runtime.Serialization; + using System.Threading; + using System.Threading.Tasks; + using Azure.Storage.Blobs; + using Azure.Storage.Blobs.Models; + using Azure.Storage.Queues.Models; + using DurableTask.Core; + using DurableTask.Core.Entities; + using DurableTask.Core.History; + using DurableTask.Core.Tracing; + 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(maxDispatchCount: 1, prefix: prefix); + + BlobServiceClient blobServiceClient = CreateBlobServiceClient(); + string containerName = $"{prefix}-instance-messages"; + 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 dispatch count of 2, which exceeds the maximum dispatch 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); + + OrchestrationState state = await client.WaitForOrchestrationAsync(instance, DefaultTimeout); + + Assert.IsNotNull(state); + Assert.AreEqual(OrchestrationStatus.Failed, state.OrchestrationStatus); + + // The orchestration-level FailureDetails is serialized into the output as "{ErrorType}: {ErrorMessage}", + // so reconstruct it from whichever the backend populated. + FailureDetails failureDetails = GetFailureDetails(state); + Assert.IsNotNull(failureDetails); + Assert.AreEqual("PoisonMessages", failureDetails.ErrorType); + StringAssert.Contains(failureDetails.ErrorMessage, EventType.ExecutionStarted.ToString()); + StringAssert.Contains(failureDetails.ErrorMessage, "maximum dispatch count"); + Assert.IsTrue(failureDetails.IsNonRetriable); + + Assert.IsTrue(await containerClient.ExistsAsync(), $"Blob container '{containerName}' should exist"); + + List blobs = await ListBlobsAsync(containerClient); + Assert.AreEqual(1, blobs.Count); + + string expectedPrefix = $"{instance.InstanceId}~{instance.ExecutionId}"; + Assert.AreEqual(expectedPrefix, blobs[0].Name.Substring(0, expectedPrefix.Length)); + + List poisonMessages = await DownloadPoisonMessagesAsync(containerClient, blobs[0].Name); + Assert.AreEqual(1, poisonMessages.Count); + Assert.IsInstanceOfType(poisonMessages[0].Event, typeof(ExecutionStartedEvent)); + var executionStartedEvent = (ExecutionStartedEvent)poisonMessages[0].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"]); + Assert.AreEqual(2, executionStartedEvent.DispatchCount); + } + finally + { + await worker.StopAsync(isForced: true); + await containerClient.DeleteIfExistsAsync(); + await service.DeleteAsync(); + } + } + + [TestMethod] + public async Task OrchestrationWithMessageEqualToMaxDispatchCount_CompletesSuccessfully() + { + string prefix = CreateUniquePrefix(); + AzureStorageOrchestrationServiceSettings settings = CreateSettings(maxDispatchCount: 2, prefix: prefix); + + BlobServiceClient blobServiceClient = CreateBlobServiceClient(); + string containerName = $"{prefix}-instance-messages"; + 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 ActivityWithPoisonMessage_Failed_AndPoisonMessageStored() + { + string prefix = CreateUniquePrefix(); + AzureStorageOrchestrationServiceSettings settings = CreateSettings(maxDispatchCount: 1, prefix: prefix); + + BlobServiceClient blobServiceClient = CreateBlobServiceClient(); + string instanceContainerName = $"{prefix}-instance-messages"; + string activityContainerName = $"{prefix}-activity-messages"; + BlobContainerClient instanceContainerClient = blobServiceClient.GetBlobContainerClient(instanceContainerName); + BlobContainerClient activityContainerClient = blobServiceClient.GetBlobContainerClient(activityContainerName); + await instanceContainerClient.DeleteIfExistsAsync(); + await activityContainerClient.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"); + + OrchestrationState state = await client.WaitForOrchestrationAsync(instance, DefaultTimeout); + + Assert.IsNotNull(state); + Assert.AreEqual(OrchestrationStatus.Failed, state.OrchestrationStatus); + + // The orchestration fails because the activity exceeded the maximum dispatch count. The poison reason + // now propagates back to the calling orchestration and is surfaced in the output. + FailureDetails failureDetails = GetFailureDetails(state); + Assert.IsNotNull(failureDetails); + StringAssert.Contains(failureDetails.ErrorMessage, "maximum dispatch count of 1"); + StringAssert.Contains(failureDetails.ErrorMessage, "dispatch count 2"); + + Assert.IsFalse( + await instanceContainerClient.ExistsAsync(), + $"Blob container '{instanceContainerName}' should not exist"); + Assert.IsTrue( + await activityContainerClient.ExistsAsync(), + $"Blob container '{activityContainerName}' should exist"); + + List blobs = await ListBlobsAsync(activityContainerClient); + Assert.AreEqual(1, blobs.Count); + + string activityName = NameVersionHelper.GetDefaultName(typeof(EchoActivity)); + string expectedPrefix = $"{activityName}~{instance.InstanceId}~{instance.ExecutionId}"; + Assert.AreEqual(expectedPrefix, blobs[0].Name.Substring(0, expectedPrefix.Length)); + + List poisonMessages = await DownloadPoisonMessagesAsync(activityContainerClient, blobs[0].Name); + Assert.AreEqual(1, poisonMessages.Count); + Assert.IsInstanceOfType(poisonMessages[0].Event, typeof(TaskScheduledEvent)); + var taskScheduledEvent = (TaskScheduledEvent)poisonMessages[0].Event; + Assert.AreEqual(activityName, taskScheduledEvent.Name); + Assert.AreEqual("[\"hello\"]", taskScheduledEvent.Input); + Assert.AreEqual(2, taskScheduledEvent.DispatchCount); + } + finally + { + await worker.StopAsync(isForced: true); + await instanceContainerClient.DeleteIfExistsAsync(); + await activityContainerClient.DeleteIfExistsAsync(); + await service.DeleteAsync(); + } + } + + [TestMethod] + public async Task ActivityWithMessageEqualToMaxDispatchCount_CompletesSuccessfully() + { + string prefix = CreateUniquePrefix(); + AzureStorageOrchestrationServiceSettings settings = CreateSettings(maxDispatchCount: 2, prefix: prefix); + + BlobServiceClient blobServiceClient = CreateBlobServiceClient(); + string activityContainerName = $"{prefix}-activity-messages"; + 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 ActivityWithInvalidWorkItem_MissingOrchestrationInstance_PoisonMessageStored() + { + string prefix = CreateUniquePrefix(); + AzureStorageOrchestrationServiceSettings settings = CreateSettings(maxDispatchCount: 5, prefix: prefix); + + BlobServiceClient blobServiceClient = CreateBlobServiceClient(); + string activityContainerName = $"{prefix}-activity-messages"; + BlobContainerClient activityContainerClient = blobServiceClient.GetBlobContainerClient(activityContainerName); + await activityContainerClient.DeleteIfExistsAsync(); + + var inner = new AzureStorageOrchestrationService(settings); + + int corruptionCount = 0; + var service = new FaultInjectingOrchestrationService(inner) + { + // Corrupt the first activity dispatch by removing its OrchestrationInstance, which routes the work + // item through the activity dispatcher's "missing OrchestrationInstance" invalid-work-item path. + CorruptActivityWorkItem = wi => + { + if (Interlocked.Increment(ref corruptionCount) == 1) + { + wi.TaskMessage.OrchestrationInstance = null; + } + }, + }; + + 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); + // After abandoning the corrupted dispatch, the activity is re-delivered and processed normally. + Assert.AreEqual(OrchestrationStatus.Completed, state.OrchestrationStatus); + + Assert.IsTrue(await activityContainerClient.ExistsAsync(), $"Blob container '{activityContainerName}' should exist"); + + List blobs = await ListBlobsAsync(activityContainerClient); + Assert.AreEqual(1, blobs.Count); + + string activityName = NameVersionHelper.GetDefaultName(typeof(EchoActivity)); + StringAssert.StartsWith(blobs[0].Name, $"{activityName}~"); + + List poisonMessages = await DownloadPoisonMessagesAsync(activityContainerClient, blobs[0].Name); + Assert.AreEqual(1, poisonMessages.Count); + // The poison message preserves the corruption that we injected. + Assert.IsNull(poisonMessages[0].OrchestrationInstance); + Assert.IsInstanceOfType(poisonMessages[0].Event, typeof(TaskScheduledEvent)); + } + finally + { + await worker.StopAsync(isForced: true); + await activityContainerClient.DeleteIfExistsAsync(); + await service.DeleteAsync(); + } + } + + [TestMethod] + public async Task ActivityWithInvalidWorkItem_NonTaskScheduledEvent_PoisonMessageStored() + { + string prefix = CreateUniquePrefix(); + AzureStorageOrchestrationServiceSettings settings = CreateSettings(maxDispatchCount: 5, prefix: prefix); + + BlobServiceClient blobServiceClient = CreateBlobServiceClient(); + string activityContainerName = $"{prefix}-activity-messages"; + BlobContainerClient activityContainerClient = blobServiceClient.GetBlobContainerClient(activityContainerName); + await activityContainerClient.DeleteIfExistsAsync(); + + var inner = new AzureStorageOrchestrationService(settings); + + int corruptionCount = 0; + var service = new FaultInjectingOrchestrationService(inner) + { + CorruptActivityWorkItem = wi => + { + if (Interlocked.Increment(ref corruptionCount) == 1) + { + // Replace the TaskScheduled event with an unrelated event type to trigger the + // "unsupported event type" invalid-work-item path in the activity dispatcher. + wi.TaskMessage.Event = new EventRaisedEvent(-1, "garbage") { Name = "notAnActivity" }; + } + }, + }; + + 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); + // After abandoning the corrupted dispatch, the activity is re-delivered and processed normally. + Assert.AreEqual(OrchestrationStatus.Completed, state.OrchestrationStatus); + + Assert.IsTrue(await activityContainerClient.ExistsAsync(), $"Blob container '{activityContainerName}' should exist"); + + List blobs = await ListBlobsAsync(activityContainerClient); + Assert.AreEqual(1, blobs.Count); + + // When the event type isn't TaskScheduled, the blob name uses an empty activity name prefix ("~"). + StringAssert.StartsWith(blobs[0].Name, "~"); + + List poisonMessages = await DownloadPoisonMessagesAsync(activityContainerClient, blobs[0].Name); + Assert.AreEqual(1, poisonMessages.Count); + Assert.IsInstanceOfType(poisonMessages[0].Event, typeof(EventRaisedEvent)); + } + finally + { + await worker.StopAsync(isForced: true); + await activityContainerClient.DeleteIfExistsAsync(); + await service.DeleteAsync(); + } + } + + [TestMethod] + public async Task ActivityWithInvalidWorkItem_MissingActivityName_CallingOrchestrationFails() + { + // When an activity work item's TaskScheduledEvent has no activity name, the activity dispatcher cannot + // dispatch it and (with poison handling enabled) stores the poison message and responds to the calling + // orchestration with a non-retriable TaskFailedEvent. This surfaces as a failed activity, which fails the + // calling orchestration. + string prefix = CreateUniquePrefix(); + AzureStorageOrchestrationServiceSettings settings = CreateSettings(maxDispatchCount: 5, prefix: prefix); + + BlobServiceClient blobServiceClient = CreateBlobServiceClient(); + string activityContainerName = $"{prefix}-activity-messages"; + BlobContainerClient activityContainerClient = blobServiceClient.GetBlobContainerClient(activityContainerName); + await activityContainerClient.DeleteIfExistsAsync(); + + var inner = new AzureStorageOrchestrationService(settings); + + var service = new FaultInjectingOrchestrationService(inner) + { + CorruptActivityWorkItem = wi => + { + // Strip the activity name from every dispatch so the message can never be dispatched. The dispatcher + // detects the missing name and fails the activity back to the orchestration. + if (wi.TaskMessage.Event is TaskScheduledEvent scheduledEvent) + { + scheduledEvent.Name = null; + } + }, + }; + + 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"); + + OrchestrationState state = await client.WaitForOrchestrationAsync(instance, DefaultTimeout); + + Assert.IsNotNull(state); + // The activity could not be dispatched (no activity name), so the calling orchestration fails with the + // poison reason propagated back to it. + Assert.AreEqual(OrchestrationStatus.Failed, state.OrchestrationStatus); + FailureDetails failureDetails = GetFailureDetails(state); + Assert.IsNotNull(failureDetails); + StringAssert.Contains(failureDetails.ErrorMessage, "does not specify an activity name"); + + // The poison message is also stored to the activity poison container before the failure is returned. + Assert.IsTrue( + await activityContainerClient.ExistsAsync(), + $"Blob container '{activityContainerName}' should exist"); + + List blobs = await ListBlobsAsync(activityContainerClient); + Assert.AreEqual(1, blobs.Count); + + // When the activity name is missing, the blob name uses an empty activity name prefix ("~"). + StringAssert.StartsWith(blobs[0].Name, "~"); + + List poisonMessages = await DownloadPoisonMessagesAsync(activityContainerClient, blobs[0].Name); + Assert.AreEqual(1, poisonMessages.Count); + Assert.IsInstanceOfType(poisonMessages[0].Event, typeof(TaskScheduledEvent)); + } + finally + { + await worker.StopAsync(isForced: true); + await activityContainerClient.DeleteIfExistsAsync(); + await service.DeleteAsync(); + } + } + + [TestMethod] + public async Task OrchestrationWithInvalidWorkItem_MissingOrchestrationInstance_FailedAndPoisonMessageStored() + { + string prefix = CreateUniquePrefix(); + AzureStorageOrchestrationServiceSettings settings = CreateSettings(maxDispatchCount: 5, prefix: prefix); + + BlobServiceClient blobServiceClient = CreateBlobServiceClient(); + string containerName = $"{prefix}-instance-messages"; + BlobContainerClient containerClient = blobServiceClient.GetBlobContainerClient(containerName); + await containerClient.DeleteIfExistsAsync(); + + var inner = new AzureStorageOrchestrationService(settings); + + int corruptionCount = 0; + var service = new FaultInjectingOrchestrationService(inner) + { + CorruptOrchestrationWorkItem = wi => + { + if (Interlocked.Increment(ref corruptionCount) == 1 && wi.NewMessages.Count > 0) + { + // Swap for a mutable list, then replace the first message's OrchestrationInstance with one that + // has an empty InstanceId. ReconcileMessagesWithState returns false with "Work item includes a + // message with no orchestration instance ID...", which routes through HandleInvalidWorkItemAsync + // and fails the orchestration with OrchestrationHistoryCorrupted. + wi.NewMessages = new List(wi.NewMessages); + + OrchestrationInstance originalInstance = wi.NewMessages[0].OrchestrationInstance; + wi.NewMessages[0].OrchestrationInstance = new OrchestrationInstance + { + InstanceId = string.Empty, + ExecutionId = string.Empty, + }; + + // Inject an extra message so the poison handler must persist the full batch. + wi.NewMessages.Add(new TaskMessage + { + OrchestrationInstance = originalInstance, + Event = new EventRaisedEvent(-1, "extra payload") { Name = "extraMarker" }, + }); + } + }, + }; + + 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); + 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.Failed, state.OrchestrationStatus); + FailureDetails failureDetails = GetFailureDetails(state); + Assert.IsNotNull(failureDetails); + Assert.AreEqual("OrchestrationHistoryCorrupted", failureDetails.ErrorType); + Assert.IsTrue(failureDetails.IsNonRetriable); + + Assert.IsTrue(await containerClient.ExistsAsync(), $"Blob container '{containerName}' should exist"); + + List blobs = await ListBlobsAsync(containerClient); + Assert.AreEqual(1, blobs.Count); + // Empty instance/execution ids sanitize to "~~{guid}". + StringAssert.StartsWith(blobs[0].Name, "~~"); + + List poisonMessages = await DownloadPoisonMessagesAsync(containerClient, blobs[0].Name); + // Both the OI-stripped ExecutionStarted message and the injected event must be persisted as poison. + Assert.AreEqual(2, poisonMessages.Count); + Assert.IsTrue(poisonMessages.Any(m => m.Event is ExecutionStartedEvent && string.IsNullOrEmpty(m.OrchestrationInstance?.InstanceId))); + Assert.IsTrue(poisonMessages.Any(m => m.Event is EventRaisedEvent raised && raised.Name == "extraMarker")); + } + finally + { + await worker.StopAsync(isForced: true); + await containerClient.DeleteIfExistsAsync(); + await service.DeleteAsync(); + } + } + + [TestMethod] + public async Task OrchestrationWithInvalidWorkItem_NoExecutionStartedEvent_FailedAndPoisonMessagesStored() + { + string prefix = CreateUniquePrefix(); + AzureStorageOrchestrationServiceSettings settings = CreateSettings(maxDispatchCount: 5, prefix: prefix); + + BlobServiceClient blobServiceClient = CreateBlobServiceClient(); + string containerName = $"{prefix}-instance-messages"; + BlobContainerClient containerClient = blobServiceClient.GetBlobContainerClient(containerName); + await containerClient.DeleteIfExistsAsync(); + + var inner = new AzureStorageOrchestrationService(settings); + + int corruptionCount = 0; + var service = new FaultInjectingOrchestrationService(inner) + { + CorruptOrchestrationWorkItem = wi => + { + if (Interlocked.Increment(ref corruptionCount) == 1 && wi.NewMessages.Count > 0) + { + // Replace the ExecutionStarted message with a non-ES message so the orchestration has no + // ExecutionStarted event either in history or in new messages. ReconcileMessagesWithState + // returns false with "Orchestration contains no ExecutionStarted event..." and fails the + // orchestration with OrchestrationHistoryCorrupted. + wi.NewMessages = new List(wi.NewMessages); + + OrchestrationInstance instance = null; + for (int i = 0; i < wi.NewMessages.Count; i++) + { + if (wi.NewMessages[i].Event is ExecutionStartedEvent) + { + instance = wi.NewMessages[i].OrchestrationInstance; + wi.NewMessages[i] = new TaskMessage + { + OrchestrationInstance = instance, + Event = new EventRaisedEvent(-1, "fake input") { Name = "fakeEvent" }, + }; + break; + } + } + + if (instance != null) + { + wi.NewMessages.Add(new TaskMessage + { + OrchestrationInstance = instance, + Event = new EventRaisedEvent(-1, "extra payload") { Name = "extraMarker" }, + }); + } + } + }, + }; + + 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); + 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.Failed, state.OrchestrationStatus); + FailureDetails failureDetails = GetFailureDetails(state); + Assert.IsNotNull(failureDetails); + Assert.AreEqual("OrchestrationHistoryCorrupted", failureDetails.ErrorType); + Assert.IsTrue(failureDetails.IsNonRetriable); + + Assert.IsTrue(await containerClient.ExistsAsync(), $"Blob container '{containerName}' should exist"); + + List blobs = await ListBlobsAsync(containerClient); + Assert.AreEqual(1, blobs.Count); + + List poisonMessages = await DownloadPoisonMessagesAsync(containerClient, blobs[0].Name); + // Both the replaced ES message and the injected rider must be persisted as poison. + Assert.AreEqual(2, poisonMessages.Count); + Assert.IsTrue(poisonMessages.Any(m => m.Event is EventRaisedEvent raised && raised.Name == "fakeEvent")); + Assert.IsTrue(poisonMessages.Any(m => m.Event is EventRaisedEvent raised && raised.Name == "extraMarker")); + } + finally + { + await worker.StopAsync(isForced: true); + await containerClient.DeleteIfExistsAsync(); + await service.DeleteAsync(); + } + } + + [TestMethod] + public async Task OrchestrationWithInvalidWorkItem_RewindWithOtherMessages_PoisonMessageStoredAndAbandoned() + { + string prefix = CreateUniquePrefix(); + AzureStorageOrchestrationServiceSettings settings = CreateSettings(maxDispatchCount: 5, prefix: prefix); + + BlobServiceClient blobServiceClient = CreateBlobServiceClient(); + string containerName = $"{prefix}-instance-messages"; + BlobContainerClient containerClient = blobServiceClient.GetBlobContainerClient(containerName); + await containerClient.DeleteIfExistsAsync(); + + var inner = new AzureStorageOrchestrationService(settings); + + int corruptionCount = 0; + var service = new FaultInjectingOrchestrationService(inner) + { + CorruptOrchestrationWorkItem = wi => + { + // Only corrupt the very first dispatch. The poison handler abandons the work item, so it is + // redelivered; we let the redelivery run normally so the orchestration can complete. + if (Interlocked.Increment(ref corruptionCount) != 1 || wi.NewMessages.Count == 0) + { + return; + } + + // Will contain the fresh ExecutionStartedEvent along with the ExecutionRewoundEvent below, which is illegal + // (the only new message should be the ExecutionRewoundEvent in this case) + wi.NewMessages = new List(wi.NewMessages); + + OrchestrationInstance instance = wi.NewMessages[0].OrchestrationInstance; + + // Pre-populate a synthetic runtime state in a terminal (Failed) status. ReconcileMessagesWithState's + // rewind poison branch requires OrchestrationStatus != Running, an ExecutionRewoundEvent in the + // batch, and NewMessages.Count > 1. The ParentTraceContext on the ExecutionStartedEvent is required + // so the dispatcher's rewind setup does not throw a NullReferenceException before the poison check. + var syntheticEs = new ExecutionStartedEvent(-1, null) + { + OrchestrationInstance = instance, + Name = NameVersionHelper.GetDefaultName(typeof(EchoOrchestration)), + Version = NameVersionHelper.GetDefaultVersion(typeof(EchoOrchestration)), + ParentTraceContext = new DistributedTraceContext("00-00000000000000000000000000000000-0000000000000000-00"), + }; + var syntheticEc = new ExecutionCompletedEvent(-1, null, OrchestrationStatus.Failed); + wi.OrchestrationRuntimeState = new OrchestrationRuntimeState(new List { syntheticEs, syntheticEc }); + + wi.NewMessages.Add(new TaskMessage + { + OrchestrationInstance = instance, + Event = new ExecutionRewoundEvent(-1, "fake rewind") + { + ParentTraceContext = new DistributedTraceContext("00-00000000000000000000000000000000-0000000000000000-00"), + }, + }); + }, + }; + + 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"); + + // The first dispatch is poisoned (blob written, work item abandoned). The redelivery runs normally + // and the orchestration completes. + OrchestrationState state = await client.WaitForOrchestrationAsync(instance, DefaultTimeout); + + Assert.IsNotNull(state); + Assert.AreEqual(OrchestrationStatus.Completed, state.OrchestrationStatus); + + Assert.IsTrue(await containerClient.ExistsAsync(), $"Blob container '{containerName}' should exist"); + + List blobs = await ListBlobsAsync(containerClient); + Assert.AreEqual(1, blobs.Count); + StringAssert.StartsWith(blobs[0].Name, $"{instance.InstanceId}~"); + + List poisonMessages = await DownloadPoisonMessagesAsync(containerClient, blobs[0].Name); + // The poisoned batch must contain the injected rewind event alongside the original ExecutionStarted message. + Assert.IsTrue(poisonMessages.Count > 1, "Expected the poisoned batch to contain the rewind event alongside other messages."); + Assert.IsTrue(poisonMessages.Any(m => m.Event is ExecutionRewoundEvent)); + Assert.IsTrue(poisonMessages.Any(m => m.Event is ExecutionStartedEvent)); + } + finally + { + await worker.StopAsync(isForced: true); + await containerClient.DeleteIfExistsAsync(); + await service.DeleteAsync(); + } + } + + [TestMethod] + public async Task OrchestrationWithDispatchExceedingMax_PoisonHandlingDisabled_CompletesSuccessfully_NoBlob() + { + // Even with MaxDispatchCount=1 and an injected transient failure that pushes DispatchCount above the limit, + // when poison handling is disabled the message must NOT be treated as poisoned, the orchestration must NOT + // fail, and no blob container should be created. + string prefix = CreateUniquePrefix(); + AzureStorageOrchestrationServiceSettings settings = CreateSettings(maxDispatchCount: 1, prefix: prefix, poisonEnabled: false); + + BlobServiceClient blobServiceClient = CreateBlobServiceClient(); + string containerName = $"{prefix}-instance-messages"; + 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 ActivityWithDispatchExceedingMax_PoisonHandlingDisabled_CompletesSuccessfully_NoBlob() + { + string prefix = CreateUniquePrefix(); + AzureStorageOrchestrationServiceSettings settings = CreateSettings(maxDispatchCount: 1, prefix: prefix, poisonEnabled: false); + + BlobServiceClient blobServiceClient = CreateBlobServiceClient(); + string activityContainerName = $"{prefix}-activity-messages"; + 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 PoisonMessageHandlerApis_PoisonHandlingDisabled_ReturnFalse() + { + // Directly invokes each IPoisonMessageHandler API on a service constructed with poison handling disabled. + // All handler APIs must return false, MaxDispatchCount must be int.MaxValue (so the dispatchers never treat + // any message as poisoned), and no blob containers should be created by these calls. + string prefix = CreateUniquePrefix(); + AzureStorageOrchestrationServiceSettings settings = CreateSettings(maxDispatchCount: 1, prefix: prefix, poisonEnabled: false); + + BlobServiceClient blobServiceClient = CreateBlobServiceClient(); + string[] containerSuffixes = { "instance-messages", "activity-messages" }; + foreach (string suffix in containerSuffixes) + { + await blobServiceClient.GetBlobContainerClient($"{prefix}-{suffix}").DeleteIfExistsAsync(); + } + + var service = new AzureStorageOrchestrationService(settings); + IPoisonMessageHandler handler = service; + + var orchInstance = new OrchestrationInstance + { + InstanceId = Guid.NewGuid().ToString("N"), + ExecutionId = Guid.NewGuid().ToString("N"), + }; + + // 1. Poison detection is effectively off (unbounded dispatch count) when handling is disabled. + Assert.AreEqual(int.MaxValue, handler.MaxDispatchCount); + + // 2. HandlePoisonEntityMessageAsync returns false. + var entityRequest = new EventRaisedEvent(eventId: -1, input: "{}") + { + Name = "op", + DispatchCount = 100, + }; + Assert.IsFalse(await handler.HandlePoisonEntityMessageAsync( + orchInstance, entityRequest, PoisonMessageReason.DeserializationError, "disabled-test")); + + // 3. HandleInvalidWorkItemAsync(TaskOrchestrationWorkItem) returns false. + var orchestrationWorkItem = new TaskOrchestrationWorkItem + { + InstanceId = orchInstance.InstanceId, + NewMessages = new List(), + OrchestrationRuntimeState = new OrchestrationRuntimeState(), + LockedUntilUtc = DateTime.UtcNow.AddMinutes(5), + }; + Assert.IsFalse(await handler.HandleInvalidWorkItemAsync( + orchestrationWorkItem, PoisonMessageReason.InvalidRuntimeState, "disabled-test", isEntity: false)); + + // 4. HandleInvalidWorkItemAsync(TaskActivityWorkItem) returns false. + var activityWorkItem = new TaskActivityWorkItem + { + Id = "act-1", + LockedUntilUtc = DateTime.UtcNow.AddMinutes(5), + TaskMessage = new TaskMessage + { + Event = new TaskScheduledEvent(eventId: -1) + { + Name = "echoActivity", + Version = "1", + DispatchCount = 100, + }, + }, + }; + Assert.IsFalse(await handler.HandleInvalidWorkItemAsync( + activityWorkItem, PoisonMessageReason.MissingActivityName, "disabled-test")); + + // 5. None of the poison blob containers should have been created. + foreach (string suffix in containerSuffixes) + { + string containerName = $"{prefix}-{suffix}"; + Assert.IsFalse( + await blobServiceClient.GetBlobContainerClient(containerName).ExistsAsync(), + $"Blob container '{containerName}' should not exist when poison handling is disabled"); + } + } + + /* + * Lower-level entity coverage: there is no in-repo way to make an orchestration call an entity, + * so we exercise HandlePoisonEntityMessageAsync directly against a real service. + */ + + [TestMethod] + public async Task EntityPoisonMessage_MalformedOpRequest_PoisonMessageStoredAndFailureResponseEnqueued() + { + // A malformed entity "op" request is poisoned; assert it is persisted to the (consolidated) instance + // poison container and that a failure response is enqueued back to the calling orchestration. + string prefix = CreateUniquePrefix(); + AzureStorageOrchestrationServiceSettings settings = CreateSettings(maxDispatchCount: 5, prefix: prefix); + // A single partition means a single control queue that we can inspect for the emitted failure response. + settings.PartitionCount = 1; + + BlobServiceClient blobServiceClient = CreateBlobServiceClient(); + string containerName = $"{prefix}-instance-messages"; + BlobContainerClient containerClient = blobServiceClient.GetBlobContainerClient(containerName); + await containerClient.DeleteIfExistsAsync(); + + var service = new AzureStorageOrchestrationService(settings); + await service.CreateAsync(recreateInstanceStore: true); + + string controlQueueName = AzureStorageOrchestrationService.GetControlQueueName(settings.TaskHubName, 0); + var controlQueueClient = new Azure.Storage.Queues.QueueClient( + TestHelpers.GetTestStorageAccountConnectionString(), controlQueueName); + + try + { + IPoisonMessageHandler handler = service; + + var entityInstance = new OrchestrationInstance + { + InstanceId = "@counter@myEntity", + ExecutionId = Guid.NewGuid().ToString("N"), + }; + + // The request id of the poisoned operation. The failure response event is named after this id so the + // calling orchestration can correlate the response with its outstanding call. + string requestId = Guid.NewGuid().ToString(); + + // An "op" (entity operation call) request whose JSON preserves the "id" and "parent" (calling + // orchestration) fields but is malformed afterwards. The strict decode fails (routing through poison + // handling), but the lenient decode can still recover the caller so a failure response can be returned. + var opEvent = new EventRaisedEvent(-1, $"{{\"op\":\"add\",\"id\":\"{requestId}\",\"parent\":\"@caller@1\",\"input\": \"not valid json") + { + Name = "op", + }; + + bool handled = await handler.HandlePoisonEntityMessageAsync( + entityInstance, opEvent, PoisonMessageReason.DeserializationError, "malformed entity op request"); + + Assert.IsTrue(handled); + + // The poison message is stored. + Assert.IsTrue(await containerClient.ExistsAsync(), $"Blob container '{containerName}' should exist"); + List blobs = await ListBlobsAsync(containerClient); + Assert.AreEqual(1, blobs.Count); + + List poisonMessages = await DownloadPoisonMessagesAsync(containerClient, blobs[0].Name); + Assert.AreEqual(1, poisonMessages.Count); + Assert.IsInstanceOfType(poisonMessages[0].Event, typeof(EventRaisedEvent)); + Assert.AreEqual("op", ((EventRaisedEvent)poisonMessages[0].Event).Name); + + // A failure response is enqueued back to the calling orchestration ("@caller@1") so it does not hang + // waiting for a response that will never come. Dequeue it and confirm it targets the caller and is + // correlated with the original request id. + QueueMessage queueMessage = null; + await TestHelpers.WaitFor( + () => + { + queueMessage = controlQueueClient.ReceiveMessage(TimeSpan.FromMinutes(1)).Value; + return queueMessage != null; + }, + TimeSpan.FromSeconds(15)); + Assert.IsNotNull(queueMessage, "Expected a failure response to be enqueued to the caller's control queue."); + + MessageData messageData = CreateMessageManager(settings).DeserializeMessageData(queueMessage.MessageText); + TaskMessage responseTaskMessage = messageData.TaskMessage; + + // The response is an EventRaisedEvent whose name is the request id, targeting the calling orchestration. + Assert.IsInstanceOfType(responseTaskMessage.Event, typeof(EventRaisedEvent)); + var responseEvent = (EventRaisedEvent)responseTaskMessage.Event; + Assert.AreEqual(requestId, responseEvent.Name); + Assert.AreEqual("@caller@1", responseTaskMessage.OrchestrationInstance.InstanceId); + + // The response payload must carry a failure so the caller observes an error rather than a success. + Assert.IsNotNull(responseEvent.Input); + StringAssert.Contains(responseEvent.Input, "EntityRequestMessageDeserializationError"); + + // No further messages should have been emitted. + QueueMessage extraMessage = controlQueueClient.ReceiveMessage(TimeSpan.FromSeconds(1)).Value; + Assert.IsNull(extraMessage, "Only a single failure response should have been emitted."); + } + finally + { + await containerClient.DeleteIfExistsAsync(); + await service.DeleteAsync(); + } + } + + [TestMethod] + public async Task EntityPoisonMessage_DispatchCountReason_StoredButReturnsFalseAndNoMessageEmitted() + { + // When an entity message is poisoned because it exceeded the maximum dispatch count (rather than because it + // could not be deserialized), the message is still archived to the poison container, but + // HandlePoisonEntityMessageAsync returns false so the dispatcher can continue processing, and it does NOT + // emit any follow-up entity message (no failure response or unlock). + string prefix = CreateUniquePrefix(); + AzureStorageOrchestrationServiceSettings settings = CreateSettings(maxDispatchCount: 5, prefix: prefix); + settings.PartitionCount = 1; + + BlobServiceClient blobServiceClient = CreateBlobServiceClient(); + string containerName = $"{prefix}-instance-messages"; + BlobContainerClient containerClient = blobServiceClient.GetBlobContainerClient(containerName); + await containerClient.DeleteIfExistsAsync(); + + var service = new AzureStorageOrchestrationService(settings); + await service.CreateAsync(recreateInstanceStore: true); + + string controlQueueName = AzureStorageOrchestrationService.GetControlQueueName(settings.TaskHubName, 0); + var controlQueueClient = new Azure.Storage.Queues.QueueClient( + TestHelpers.GetTestStorageAccountConnectionString(), controlQueueName); + + try + { + IPoisonMessageHandler handler = service; + + var entityInstance = new OrchestrationInstance + { + InstanceId = "@counter@myEntity", + ExecutionId = Guid.NewGuid().ToString("N"), + }; + + // A well-formed release message that is poisoned only because it exceeded the dispatch count. + var releaseEvent = new EventRaisedEvent(-1, "{\"parent\":\"@caller@1\",\"id\":\"fix\"}") + { + Name = "release", + }; + + bool handled = await handler.HandlePoisonEntityMessageAsync( + entityInstance, releaseEvent, PoisonMessageReason.DispatchCount, "dispatch count exceeded"); + + // The dispatch-count reason must return false (let the dispatcher continue) ... + Assert.IsFalse(handled); + + // ... the message is still archived to the poison container ... + Assert.IsTrue( + await containerClient.ExistsAsync(), + $"Blob container '{containerName}' should exist"); + List blobs = await ListBlobsAsync(containerClient); + Assert.AreEqual(1, blobs.Count); + List poisonMessages = await DownloadPoisonMessagesAsync(containerClient, blobs[0].Name); + Assert.AreEqual(1, poisonMessages.Count); + Assert.IsInstanceOfType(poisonMessages[0].Event, typeof(EventRaisedEvent)); + Assert.AreEqual("release", ((EventRaisedEvent)poisonMessages[0].Event).Name); + + // ... but no follow-up entity message (failure response or unlock) should have been emitted. + QueueMessage extraMessage = controlQueueClient.ReceiveMessage(TimeSpan.FromSeconds(1)).Value; + Assert.IsNull(extraMessage, "No message should be emitted for a dispatch-count poison message."); + } + finally + { + await containerClient.DeleteIfExistsAsync(); + await service.DeleteAsync(); + } + } + + [TestMethod] + public async Task EntityPoisonMessage_MalformedReleaseRequest_PoisonMessageStoredAndUnlockEmitted() + { + string prefix = CreateUniquePrefix(); + AzureStorageOrchestrationServiceSettings settings = CreateSettings(maxDispatchCount: 5, prefix: prefix); + // A single partition means a single control queue that we can inspect for the emitted unlock message. + settings.PartitionCount = 1; + + BlobServiceClient blobServiceClient = CreateBlobServiceClient(); + string containerName = $"{prefix}-instance-messages"; + BlobContainerClient containerClient = blobServiceClient.GetBlobContainerClient(containerName); + await containerClient.DeleteIfExistsAsync(); + + var service = new AzureStorageOrchestrationService(settings); + await service.CreateAsync(recreateInstanceStore: true); + + string controlQueueName = AzureStorageOrchestrationService.GetControlQueueName(settings.TaskHubName, 0); + var controlQueueClient = new Azure.Storage.Queues.QueueClient( + TestHelpers.GetTestStorageAccountConnectionString(), controlQueueName); + + try + { + IPoisonMessageHandler handler = service; + + var entityInstance = new OrchestrationInstance + { + InstanceId = "@counter@myEntity", + ExecutionId = Guid.NewGuid().ToString("N"), + }; + + // A "release" (lock release) message whose JSON preserves the "parent" (lock owner) field but is + // malformed afterwards. The strict decode fails (routing through poison handling), but the lenient + // decode can still recover the lock owner so a fresh unlock can be emitted to the entity. + var releaseEvent = new EventRaisedEvent(-1, "{\"parent\":\"@caller@1\",\"id\":\"fix\",\"lockset\": \"not valid json") + { + Name = "release", + }; + + bool handled = await handler.HandlePoisonEntityMessageAsync( + entityInstance, releaseEvent, PoisonMessageReason.DeserializationError, "malformed entity release"); + + Assert.IsTrue(handled); + + // The poison message is stored. + Assert.IsTrue(await containerClient.ExistsAsync(), $"Blob container '{containerName}' should exist"); + List blobs = await ListBlobsAsync(containerClient); + Assert.AreEqual(1, blobs.Count); + + List poisonMessages = await DownloadPoisonMessagesAsync(containerClient, blobs[0].Name); + Assert.AreEqual(1, poisonMessages.Count); + Assert.IsInstanceOfType(poisonMessages[0].Event, typeof(EventRaisedEvent)); + Assert.AreEqual("release", ((EventRaisedEvent)poisonMessages[0].Event).Name); + + // A fresh unlock (release) message is emitted to the entity's control queue so the entity does not + // remain locked forever by the failed caller. Dequeue it and confirm it is a well-formed release that + // targets the entity and recovers the original lock owner ("@caller@1"). + QueueMessage queueMessage = null; + await TestHelpers.WaitFor( + () => + { + queueMessage = controlQueueClient.ReceiveMessage(TimeSpan.FromMinutes(1)).Value; + return queueMessage != null; + }, + TimeSpan.FromSeconds(15)); + Assert.IsNotNull(queueMessage, "Expected an unlock message to be emitted to the entity's control queue."); + + MessageData messageData = CreateMessageManager(settings).DeserializeMessageData(queueMessage.MessageText); + TaskMessage unlockTaskMessage = messageData.TaskMessage; + + // The emitted message must be an entity "release" event targeting the entity instance. + Assert.IsInstanceOfType(unlockTaskMessage.Event, typeof(EventRaisedEvent)); + var unlockEvent = (EventRaisedEvent)unlockTaskMessage.Event; + Assert.AreEqual("release", unlockEvent.Name); + Assert.AreEqual(entityInstance.InstanceId, unlockTaskMessage.OrchestrationInstance.InstanceId); + + // The recreated release must recover the original lock owner so the correct caller's lock is released. + Assert.IsNotNull(unlockEvent.Input); + StringAssert.Contains(unlockEvent.Input, "@caller@1"); + + // No further messages should have been emitted. + QueueMessage extraMessage = controlQueueClient.ReceiveMessage(TimeSpan.FromSeconds(1)).Value; + Assert.IsNull(extraMessage, "Only a single unlock message should have been emitted."); + } + finally + { + await containerClient.DeleteIfExistsAsync(); + await service.DeleteAsync(); + } + } + + static AzureStorageOrchestrationServiceSettings CreateSettings( + int maxDispatchCount, + 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.MaxDispatchCount = maxDispatchCount; + settings.PoisonMessageStorageContainerNamePrefix = prefix; + + return settings; + } + + static BlobServiceClient CreateBlobServiceClient() + { + return new BlobServiceClient(TestHelpers.GetTestStorageAccountConnectionString()); + } + + // 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) + { + 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"); + + List poisonMessages = JsonConvert.DeserializeObject>( + blobContent, + new JsonSerializerSettings { TypeNameHandling = TypeNameHandling.Auto }); + + Assert.IsNotNull(poisonMessages); + return poisonMessages; + } + + 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, IPoisonMessageHandler + { + 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; + + IPoisonMessageHandler InnerPoisonHandler => 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); + + // ---- IPoisonMessageHandler ---- + + public int MaxDispatchCount => this.InnerPoisonHandler.MaxDispatchCount; + + public Task HandlePoisonEntityMessageAsync(OrchestrationInstance entityInstance, HistoryEvent historyEvent, PoisonMessageReason reason, string details) => + this.InnerPoisonHandler.HandlePoisonEntityMessageAsync(entityInstance, historyEvent, reason, details); + + public Task HandleInvalidWorkItemAsync(TaskOrchestrationWorkItem workItem, PoisonMessageReason reason, string details, bool isEntity) => + this.InnerPoisonHandler.HandleInvalidWorkItemAsync(workItem, reason, details, isEntity); + + public Task HandleInvalidWorkItemAsync(TaskActivityWorkItem workItem, PoisonMessageReason reason, string details) => + this.InnerPoisonHandler.HandleInvalidWorkItemAsync(workItem, reason, details); + } + } +} diff --git a/src/DurableTask.AzureStorage/AzureStorageOrchestrationService.cs b/src/DurableTask.AzureStorage/AzureStorageOrchestrationService.cs index 74798cb45..32cd2db42 100644 --- a/src/DurableTask.AzureStorage/AzureStorageOrchestrationService.cs +++ b/src/DurableTask.AzureStorage/AzureStorageOrchestrationService.cs @@ -22,8 +22,6 @@ namespace DurableTask.AzureStorage using System.Threading; using System.Threading.Tasks; using Azure; - using Azure.Data.Tables; - using Azure.Storage.Blobs.Models; using Azure.Storage.Queues.Models; using DurableTask.AzureStorage.Messaging; using DurableTask.AzureStorage.Monitoring; @@ -46,7 +44,8 @@ public sealed class AzureStorageOrchestrationService : IDisposable, IOrchestrationServiceQueryClient, IOrchestrationServicePurgeClient, - IEntityOrchestrationService + IEntityOrchestrationService, + IPoisonMessageHandler { static readonly HistoryEvent[] EmptyHistoryEventList = new HistoryEvent[0]; @@ -73,6 +72,9 @@ public sealed class AzureStorageOrchestrationService : readonly IPartitionManager partitionManager; readonly object hubCreationLock; + readonly BlobContainer instancePoisonMessageContainer; + readonly BlobContainer activityPoisonMessageContainer; + bool isStarted; Task statsLoop; CancellationTokenSource shutdownSource; @@ -186,6 +188,16 @@ public AzureStorageOrchestrationService(AzureStorageOrchestrationServiceSettings this.settings.TaskHubName.ToLowerInvariant() + "-applease", this.settings.TaskHubName.ToLowerInvariant() + "-appleaseinfo", this.settings.AppLeaseOptions); + + if (this.settings.IsPoisonMessageStorageEnabled) + { + string prefix = string.IsNullOrEmpty(this.settings.PoisonMessageStorageContainerNamePrefix) + ? "durable-task-poison" + : this.settings.PoisonMessageStorageContainerNamePrefix; + + this.instancePoisonMessageContainer = this.azureStorageClient.GetBlobContainerReference($"{prefix}-instance-messages"); + this.activityPoisonMessageContainer = this.azureStorageClient.GetBlobContainerReference($"{prefix}-activity-messages"); + } } internal string WorkerId => this.settings.WorkerId; @@ -796,7 +808,11 @@ async Task LockNextTaskOrchestrationWorkItemAsync(boo { InstanceId = session.Instance.InstanceId, LockedUntilUtc = session.CurrentMessageBatch.Min(msg => msg.OriginalQueueMessage.NextVisibleOn.Value.UtcDateTime), - NewMessages = session.CurrentMessageBatch.Select(m => m.TaskMessage).ToList(), + NewMessages = session.CurrentMessageBatch.Select(m => + { + m.TaskMessage.Event.DispatchCount = (int)m.OriginalQueueMessage.DequeueCount; + return m.TaskMessage; + }).ToList(), OrchestrationRuntimeState = session.RuntimeState, Session = this.settings.ExtendedSessionsEnabled ? session : null, TraceContext = currentRequestTraceContext, @@ -1278,7 +1294,15 @@ await this.CommitOutboundQueueMessages( throw; } - // Finally, delete the messages which triggered this orchestration execution. This is the final commit. + // Finally, store any poison messages and delete the messages which triggered this orchestration execution. This is the final commit. + if (this.settings.IsPoisonMessageStorageEnabled) + { + await this.StorePoisonMessagesAsync( + runtimeState.OrchestrationInstance, + workItem.NewMessages.Where(m => m.Event.DispatchCount > this.settings.MaxDispatchCount), + this.instancePoisonMessageContainer); + } + await this.DeleteMessageBatchAsync(session, session.CurrentMessageBatch); } @@ -1585,6 +1609,8 @@ public async Task LockNextTaskActivityWorkItem( this.stats.ActiveActivityExecutions.Increment(); + session.MessageData.TaskMessage.Event.DispatchCount = (int)message.OriginalQueueMessage.DequeueCount; + return new TaskActivityWorkItem { Id = message.Id, @@ -1630,6 +1656,23 @@ public async Task CompleteTaskActivityWorkItemAsync(TaskActivityWorkItem workIte }); // Next, delete the work item queue message. This must come after enqueuing the response message. + // Before deleting, persist the message to poison storage if it has exceeded the maximum dispatch count. + if (this.settings.IsPoisonMessageStorageEnabled + && workItem.TaskMessage.Event.DispatchCount > this.settings.MaxDispatchCount) + { + string activityName = workItem.TaskMessage.Event is TaskScheduledEvent scheduledEvent + ? scheduledEvent.Name ?? string.Empty + : string.Empty; + + // We must wait for this to finish before deleting the message so that the poison message is persisted + // in storage before the message that triggered it is removed from the queue. + await this.StorePoisonMessagesAsync( + workItem.TaskMessage.OrchestrationInstance, + new[] { workItem.TaskMessage }, + this.activityPoisonMessageContainer, + blobNamePrefix: $"{activityName}~"); + } + await this.workItemQueue.DeleteMessageAsync(session.MessageData, session); if (this.activeActivitySessions.TryRemove(workItem.Id, out _)) @@ -2230,6 +2273,403 @@ private static OrchestrationQueryResult ConvertFrom(DurableStatusQueryResult sta return new OrchestrationQueryResult(results, statusContext.ContinuationToken); } + #region Poison message handling + + /// + int IPoisonMessageHandler.MaxDispatchCount => + this.settings.IsPoisonMessageStorageEnabled ? this.settings.MaxDispatchCount : int.MaxValue; + + /// + async Task IPoisonMessageHandler.HandlePoisonEntityMessageAsync( + OrchestrationInstance? entityInstance, + HistoryEvent historyEvent, + PoisonMessageReason reason, + string details) + { + if (!this.settings.IsPoisonMessageStorageEnabled || + !await this.StorePoisonMessagesAsync( + entityInstance, + new[] { new TaskMessage { OrchestrationInstance = entityInstance, Event = historyEvent } }, + this.instancePoisonMessageContainer)) + { + return false; + } + + if (historyEvent is EventRaisedEvent eventRaisedEvent) + { + // For a self-continue or lock release message that we were able to deserialize (i.e. it is poisoned because it + // exceeded the maximum dispatch count), we don't want to leave the entity stuck, so we return false + // to allow the dispatcher to continue processing + if (reason == PoisonMessageReason.DispatchCount) + { + return false; + } + // This is an entity request message, either a call, lock, or signal. + // For the first two cases we want to try and return a failure to the calling orchestration if possible. + else if (eventRaisedEvent.Name == "op") + { + EntityMessageEvent? failureResponse = ClientEntityHelpers.TryCreateEntityOperationFailedResponse( + eventRaisedEvent.Input, + "EntityRequestMessageDeserializationError", + details); + + await this.TrySendEntityMessageAsync(entityInstance, failureResponse); + } + // This is a lock release message we were not able to deserialize. To avoid leaving the entity locked forever, + // we will attempt a more lenient deserialization to see if we can extract the locking parent instance ID, and + // send the unlock message to the entity again + else if (eventRaisedEvent.Name == "release") + { + string? parentInstanceId = null; + EntityMessageEvent? unlockEvent = entityInstance != null + ? ClientEntityHelpers.TryRecreateEntityUnlock(eventRaisedEvent.Input, entityInstance, out parentInstanceId) + : null; + + await this.TrySendEntityMessageAsync(new OrchestrationInstance { InstanceId = parentInstanceId }, unlockEvent); + } + + } + else + { + // The only type of poison message we could ever receive via this API should be an EventRaisedEvent + this.settings.Logger.GeneralError( + this.azureStorageClient.QueueAccountName, + this.settings.TaskHubName, + $"Received unexpected poison message with event type {historyEvent.EventType}. " + + $"Only entity messages of type {EventType.EventRaised} are expected to be received."); + } + + return true; + } + + async Task TrySendEntityMessageAsync(OrchestrationInstance? sourceInstance, EntityMessageEvent? messageEvent) + { + if (messageEvent == null) + { + return; + } + + try + { + EntityMessageEvent eventToSend = messageEvent.Value; + string targetInstanceId = eventToSend.TargetInstance.InstanceId; + ControlQueue? targetControlQueue = await this.GetControlQueueAsync(targetInstanceId); + if (targetControlQueue != null) + { + await targetControlQueue.AddMessageAsync( + eventToSend.AsTaskMessage(), + sourceInstance ?? EmptySourceInstance); + } + } + catch (Exception e) + { + this.settings.Logger.GeneralError( + this.azureStorageClient.QueueAccountName, + this.settings.TaskHubName, + $"Failed to send entity message during poison entity message handling for instance " + + $"InstanceId={sourceInstance?.InstanceId ?? string.Empty}. Error: {e}"); + } + } + + /// + async Task IPoisonMessageHandler.HandleInvalidWorkItemAsync(TaskOrchestrationWorkItem workItem, PoisonMessageReason reason, string details, bool isEntity) + { + if (!this.settings.IsPoisonMessageStorageEnabled) + { + return false; + } + + OrchestrationInstance? orchestrationInstance = workItem.OrchestrationRuntimeState?.OrchestrationInstance; + + if (!await this.StorePoisonMessagesAsync(orchestrationInstance, workItem.NewMessages, this.instancePoisonMessageContainer)) + { + return false; + } + + // There must have been a deserialization error with the ExecutionStartedEvent, or the work item is somehow + // otherwise invalid (i.e. missing an ExecutionStartedEvent). + // We make a best effort attempt to send a failure response for any entity calls included in the work item, + // but do not complete the work item since that would cause deletion of the other entity messages included in it and we + // have no real way to mark an entity as "failed". + // These messages will presumably continue to be retried since there is no completion to delete them, so all subsequent + // attempts will land in poison storage as well. + if (isEntity) + { + foreach (TaskMessage message in workItem.NewMessages) + { + if (message.Event is EventRaisedEvent eventRaisedEvent + && eventRaisedEvent.Name != null + && eventRaisedEvent.Name.StartsWith("op", StringComparison.Ordinal)) + { + EntityMessageEvent? failureResponse = ClientEntityHelpers.TryCreateEntityOperationFailedResponse( + eventRaisedEvent.Input, + "SchedulerStateDeserializationError", + details); + + await this.TrySendEntityMessageAsync(orchestrationInstance, failureResponse); + } + } + } + // If this was related to rewind, then presumably the orchestration is already in a terminal state so + // there is not much we can do. We just store the messages in poison storage. + // They will probably continue to be retried since there is no completion to delete them, so all subsequent + // attempts will land in poison storage as well. + else if (reason != PoisonMessageReason.InvalidRewindRequest + && await this.TryCompleteInvalidWorkItemAsync(workItem, details)) + { + return true; + } + + // We have no sensible way to complete the work item, or there was an error when attempting to do so, so we + // just abandon it. The messages have already been stored in poison storage. + try + { + await this.AbandonTaskOrchestrationWorkItemAsync(workItem); + } + catch (Exception ex) + { + this.settings.Logger.GeneralError( + this.azureStorageClient.QueueAccountName, + this.settings.TaskHubName, + $"Failed to abandon invalid instance work item. " + + $"InstanceId={workItem?.InstanceId ?? string.Empty}. " + + $"ExecutionId={orchestrationInstance?.ExecutionId ?? string.Empty}. " + + $"Error: {ex}"); + } + + return true; + } + + async Task TryCompleteInvalidWorkItemAsync(TaskOrchestrationWorkItem workItem, string reason) + { + try + { + if (!this.orchestrationSessionManager.TryGetExistingSession(workItem.InstanceId, out OrchestrationSession session)) + { + return false; + } + + session.StartNewLogicalTraceScope(); + + // Best effort: fail the orchestration by appending a failed ExecutionCompleted event to its history + // and updating the instance metadata to a Failed terminal status. + OrchestrationRuntimeState runtimeState = session.RuntimeState; + string instanceId = workItem.InstanceId; + string? executionId = runtimeState.OrchestrationInstance?.ExecutionId ?? session.Instance?.ExecutionId; + + if (executionId != null) + { + var failureDetails = new FailureDetails( + errorType: "OrchestrationHistoryCorrupted", + errorMessage: reason, + stackTrace: null, + innerFailure: null, + isNonRetriable: true); + + runtimeState.AddEvent(new ExecutionCompletedEvent( + eventId: -1, + result: null, + orchestrationStatus: OrchestrationStatus.Failed, + failureDetails: failureDetails) + { + Timestamp = DateTime.UtcNow, + }); + + // Commit the updated history and instance table metadata (RuntimeStatus = Failed). + await this.trackingStore.UpdateStateAsync( + workItem.OrchestrationRuntimeState, + workItem.OrchestrationRuntimeState, + instanceId, + executionId, + session.ETags, + session.TrackingStoreContext); + + session.UpdateRuntimeState(runtimeState); + } + + // Delete the messages that triggered this invalid work item so they are not retried forever. + await this.DeleteMessageBatchAsync(session, session.CurrentMessageBatch); + return true; + } + catch (Exception ex) + { + this.settings.Logger.GeneralError( + this.azureStorageClient.QueueAccountName, + this.settings.TaskHubName, + $"Failed to complete invalid work item. InstanceId={workItem.InstanceId}. Error: {ex}"); + return false; + } + } + + + /// + async Task IPoisonMessageHandler.HandleInvalidWorkItemAsync(TaskActivityWorkItem workItem, PoisonMessageReason reason, string details) + { + if (!this.settings.IsPoisonMessageStorageEnabled) + { + return false; + } + + TaskMessage taskMessage = workItem.TaskMessage; + string activityName = taskMessage.Event is TaskScheduledEvent scheduledEvent + ? scheduledEvent.Name ?? string.Empty + : string.Empty; + + if (!await this.StorePoisonMessagesAsync( + taskMessage.OrchestrationInstance, + new[] { taskMessage }, + this.activityPoisonMessageContainer, + $"{activityName}~")) + { + return false; + } + + // If missing an activity name, the dispatcher will fail the corresponding orchestration so we do not want to abandon the + // activity work item here + if (reason != PoisonMessageReason.MissingActivityName) + { + try + { + await this.AbandonTaskActivityWorkItemAsync(workItem); + } + catch (Exception ex) + { + this.settings.Logger.GeneralError( + this.azureStorageClient.QueueAccountName, + this.settings.TaskHubName, + $"Failed to abandon invalid activity work item. " + + $"InstanceId={taskMessage.OrchestrationInstance?.InstanceId ?? string.Empty} " + + $"ExecutionId={taskMessage.OrchestrationInstance?.ExecutionId ?? string.Empty}. " + + $"Error: {ex}"); + } + } + + return true; + } + + async Task StorePoisonMessagesAsync( + OrchestrationInstance? orchestrationInstance, + IEnumerable poisonMessages, + BlobContainer container, + string blobNamePrefix = "") + { + if (poisonMessages.Count() > 0) + { + try + { + // Replace any invalid characters with a dash + string sanitizedInstanceId = SanitizeString(orchestrationInstance?.InstanceId, '-'); + blobNamePrefix += $"{sanitizedInstanceId}~{orchestrationInstance?.ExecutionId ?? string.Empty}"; + + // Blob name length limit is 1024 characters and we attach an extra character (~) and GUID (32) at the end + // From https://learn.microsoft.com/en-us/rest/api/storageservices/naming-and-referencing-containers--blobs--and-metadata?#blob-names + const int MaxPrefixLength = 1024 - 32 - 1; + if (blobNamePrefix.Length > MaxPrefixLength) + { + blobNamePrefix = blobNamePrefix.Substring(0, MaxPrefixLength); + } + string blobName = $"{blobNamePrefix}~{Guid.NewGuid():N}"; + + await container.CreateIfNotExistsAsync(); + + string serializedPoisonMessages = JsonConvert.SerializeObject( + poisonMessages, + new JsonSerializerSettings { TypeNameHandling = TypeNameHandling.Auto }); + + Blob blob = container.GetBlobReference(blobName); + await blob.UploadTextAsync(serializedPoisonMessages); + + this.settings.Logger.GeneralWarning( + this.azureStorageClient.BlobAccountName, + this.settings.TaskHubName, + $"Stored {poisonMessages.Count()} poison message(s) for instance InstanceId='{orchestrationInstance?.InstanceId ?? string.Empty}' " + + $"ExecutionId='{orchestrationInstance?.ExecutionId ?? string.Empty}' in blob '{blobName}'."); + } + catch (Exception e) + { + this.settings.Logger.GeneralError( + this.azureStorageClient.BlobAccountName, + this.settings.TaskHubName, + $"Error when attempting to store poison messages for instance " + + $"InstanceId={orchestrationInstance?.InstanceId ?? string.Empty} " + + $"ExecutionId={orchestrationInstance?.ExecutionId ?? string.Empty} " + + $"Error: {e}"); + + return false; + } + } + return true; + } + + 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 + { + scalar = c; + } + + if (IsInvalidCodePoint(scalar)) + { + sb.Append(replacement); + } + else + { + sb.Append(char.ConvertFromUtf32(scalar)); + } + } + + return sb.ToString(); + } + + #endregion + class PendingMessageBatch { public string? OrchestrationInstanceId { get; set; } diff --git a/src/DurableTask.AzureStorage/AzureStorageOrchestrationServiceSettings.cs b/src/DurableTask.AzureStorage/AzureStorageOrchestrationServiceSettings.cs index e614da92f..538240197 100644 --- a/src/DurableTask.AzureStorage/AzureStorageOrchestrationServiceSettings.cs +++ b/src/DurableTask.AzureStorage/AzureStorageOrchestrationServiceSettings.cs @@ -14,6 +14,7 @@ namespace DurableTask.AzureStorage { using System; + using System.Collections.Generic; using System.Runtime.Serialization; using Azure.Data.Tables; using DurableTask.AzureStorage.Logging; @@ -295,6 +296,33 @@ internal LogHelper Logger /// public QueueClientMessageEncoding QueueClientMessageEncoding { get; set; } = QueueClientMessageEncoding.UTF8; + /// + /// Gets or sets a flag whether poison storage is enabled. + /// + /// + /// If enabled, orchestration, entity, and activity messages that have been dispatched for processing + /// more than times will be "failed" and moved to the poison storage. + /// + /// + public bool IsPoisonMessageStorageEnabled { get; set; } = false; + + /// + /// Gets or sets the amount of times a message is dispatched for processing before it is considered "poisoned" + /// and moved to the poison storage. The default value is 5. + /// + /// + /// This setting is applicable when is set to true. + /// + public int MaxDispatchCount { get; set; } = 5; + + /// + /// Gets or sets the Azure Blob Storage container name prefix to use for poison message storage. + /// Two containers will be created with this prefix in the format "{prefix}-instance-messages" (used for both + /// orchestration and entity messages) and "{prefix}-activity-messages". + /// If not specified, the default value "durable-task-poison" will be used. + /// + public string? PoisonMessageStorageContainerNamePrefix { get; set; } = "durable-task-poison"; + /// /// When true, an etag is used when attempting to make instance table updates upon completing an orchestration work item. /// diff --git a/src/DurableTask.Core/Entities/ClientEntityHelpers.cs b/src/DurableTask.Core/Entities/ClientEntityHelpers.cs index 7151fdbe4..2e55f7a93 100644 --- a/src/DurableTask.Core/Entities/ClientEntityHelpers.cs +++ b/src/DurableTask.Core/Entities/ClientEntityHelpers.cs @@ -91,6 +91,135 @@ public static EntityMessageEvent EmitUnlockForOrphanedLock(OrchestrationInstance return new EntityMessageEvent(EntityMessageEventNames.ReleaseMessageEventName, message, targetInstance); } + /// + /// Attempts to create a failure response event that can be sent back to the orchestration that called an entity, + /// in the case where the entity operation request could not be processed. + /// + /// + /// A response is only produced for genuine two-way entity operation calls, which include lock requests and calls. + /// Signals, releases, and self-continue messages have no caller awaiting a response, so this method returns + /// for those (and for any request that cannot be decoded well enough to identify the + /// caller). + /// + /// The serialized input of the poison entity request (the Input of the + /// ). + /// The error type to record in the failure response. + /// The human-readable reason for the failure. + /// An targeting the calling instance, or if no + /// caller can or should be notified. + public static EntityMessageEvent? TryCreateEntityOperationFailedResponse(string? entityRequestInput, string errorType, string errorMessage) + { + if (string.IsNullOrEmpty(entityRequestInput)) + { + return null; + } + + RequestMessage? requestMessage = TryDecodeEntityMessage(entityRequestInput!); + + // If we cannot recover the caller info, there is no one to notify. We also require a valid request id, + // since the response event is named after it so the caller can correlate the response with its outstanding + // call. A default (empty) id means the id could not be recovered from the (malformed) request. + if (requestMessage == null + || string.IsNullOrEmpty(requestMessage.ParentInstanceId) + || requestMessage.Id == Guid.Empty) + { + return null; + } + + var responseMessage = new ResponseMessage() + { + FailureDetails = new FailureDetails( + errorType: errorType, + errorMessage: errorMessage, + stackTrace: null, + innerFailure: null, + isNonRetriable: true), + }; + + if (requestMessage.IsLockRequest) + { + responseMessage.Result = ResponseMessage.LockAcquisitionCompletion; + } + + var destination = new OrchestrationInstance() + { + InstanceId = requestMessage.ParentInstanceId!, + ExecutionId = requestMessage.ParentExecutionId, + }; + + return new EntityMessageEvent( + EntityMessageEventNames.ResponseMessageEventName(requestMessage.Id), + responseMessage, + destination); + } + + /// + /// Attempts to create a lock release event that can be sent to an entity to release an orphaned lock, in the + /// case where an incoming lock release message could not be processed. + /// + /// + /// This method attempts to decode the original release message (falling back to a lenient decode) in order to + /// recover the instance id of the lock owner. If that id cannot be recovered, this method returns + /// since there is no way to know which lock to release. + /// + /// The serialized input of the original release message. + /// The entity instance whose lock should be released. + /// The parent instance ID of the lock owner, or if it cannot be determined. + /// An targeting the entity, or if the lock + /// owner cannot be identified. + public static EntityMessageEvent? TryRecreateEntityUnlock(string? releaseMessageInput, OrchestrationInstance entityInstance, out string? parentInstanceId) + { + parentInstanceId = null; + + if (string.IsNullOrEmpty(releaseMessageInput)) + { + return null; + } + + ReleaseMessage? releaseMessage = TryDecodeEntityMessage(releaseMessageInput!); + + // If we cannot recover the lock owner, there is no way to know which lock to release. + if (releaseMessage == null || string.IsNullOrEmpty(releaseMessage.ParentInstanceId)) + { + return null; + } + + parentInstanceId = releaseMessage.ParentInstanceId; + return EmitUnlockForOrphanedLock(entityInstance, releaseMessage.ParentInstanceId!); + } + + static T? TryDecodeEntityMessage(string input) where T : EntityMessage, new() + { + // First attempt a strict deserialization. + try + { + var message = new T(); + JsonConvert.PopulateObject(input, message, Serializer.InternalSerializerSettings); + return message; + } + catch (Exception) + { + // Best-effort recovery: parse with an error-handling reader so any fields before the malformed + // segment (notably the caller/lock-owner IDs) remain available. + } + + try + { + var lenientSettings = new JsonSerializerSettings + { + Error = (_, args) => args.ErrorContext.Handled = true, + }; + + var message = new T(); + JsonConvert.PopulateObject(input, message, lenientSettings); + return message; + } + catch (Exception) + { + return null; + } + } + /// /// Extracts the user-defined entity state from the serialized scheduler state. The result is the serialized state, /// or null if the entity has no state. diff --git a/src/DurableTask.Core/Entities/EventFormat/RequestMessage.cs b/src/DurableTask.Core/Entities/EventFormat/RequestMessage.cs index ce355ab12..d7186b5cc 100644 --- a/src/DurableTask.Core/Entities/EventFormat/RequestMessage.cs +++ b/src/DurableTask.Core/Entities/EventFormat/RequestMessage.cs @@ -122,6 +122,11 @@ internal class RequestMessage : EntityMessage /// public string? ClientSpanId { get; set; } + /// + /// The number of times this request has been dispatched. + /// + public int DispatchCount { get; set; } + /// public override string GetShortDescription() { diff --git a/src/DurableTask.Core/Entities/OrchestrationEntityContext.cs b/src/DurableTask.Core/Entities/OrchestrationEntityContext.cs index 323cba441..fb0c7ee17 100644 --- a/src/DurableTask.Core/Entities/OrchestrationEntityContext.cs +++ b/src/DurableTask.Core/Entities/OrchestrationEntityContext.cs @@ -341,6 +341,16 @@ public void CompleteAcquire(OperationResult result, Guid criticalSectionId) this.lockAcquisitionPending = false; } + /// + /// Called when the entity lock acquisition fails. + /// + public void AbandonAcquire() + { + this.criticalSectionLocks = null; + this.criticalSectionId = null; + this.lockAcquisitionPending = false; + } + internal void AdjustOutgoingMessage(string instanceId, RequestMessage requestMessage, DateTime? cappedTime, out string eventName) { if (cappedTime.HasValue) 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/History/HistoryEvent.cs b/src/DurableTask.Core/History/HistoryEvent.cs index 8e6a8f316..c99510b2c 100644 --- a/src/DurableTask.Core/History/HistoryEvent.cs +++ b/src/DurableTask.Core/History/HistoryEvent.cs @@ -89,6 +89,12 @@ protected HistoryEvent(int eventId) [DataMember] public virtual EventType EventType { get; private set; } + /// + /// Gets or sets the number of times this event has been dispatched. + /// + [DataMember(EmitDefaultValue = false)] + public int DispatchCount { get; set; } + /// /// Implementation for . /// diff --git a/src/DurableTask.Core/IPoisonMessageHandler.cs b/src/DurableTask.Core/IPoisonMessageHandler.cs new file mode 100644 index 000000000..b37114a03 --- /dev/null +++ b/src/DurableTask.Core/IPoisonMessageHandler.cs @@ -0,0 +1,126 @@ +// ---------------------------------------------------------------------------------- +// 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. +// ---------------------------------------------------------------------------------- +#nullable enable +namespace DurableTask.Core +{ + using System.Threading.Tasks; + using DurableTask.Core.History; + + /// + /// Categorizes why a message or work item was considered "poisoned" so that an + /// can decide how to handle it without parsing human-readable strings. + /// + public enum PoisonMessageReason + { + /// + /// A message or event could not be deserialized + /// + DeserializationError, + + /// + /// A message was dispatched for processing more than + /// times and is therefore considered "poisoned". + /// + DispatchCount, + + /// + /// A message or work item did not contain the orchestration instance information required to process it. + /// + MissingOrchestrationInstanceId, + + /// + /// The orchestration runtime state reconstructed from history was invalid (for example, corrupted or + /// partially deleted history) and could not be processed. + /// + InvalidRuntimeState, + + /// + /// The orchestration history did not contain the required and + /// did not receive one as part of its new messages. + /// + MissingExecutionStartedEvent, + + /// + /// A rewind request was invalid, for example because additional messages were delivered alongside the rewind + /// request to an instance attempting to rewind from a terminal state. + /// + InvalidRewindRequest, + + /// + /// A work item contained an event whose type is not supported by the dispatcher that received it. + /// + WrongEventType, + + /// + /// An activity work item's did not specify an activity name, so the + /// activity could not be dispatched. + /// + MissingActivityName, + } + + /// + /// Provides extensibility points for detecting and handling "poison" messages and invalid work items + /// in the task dispatchers. + /// + public interface IPoisonMessageHandler + { + /// + /// The maximum dispatch count after which a message should be considered "poisoned" if it is dispatched again. + /// + public int MaxDispatchCount { get; } + + /// + /// Invoked to handle a poison entity message in the case that it cannot necessarily + /// be "failed" by the dispatchers, so the must + /// decide what to do. + /// + /// + /// If this method returns false, the dispatcher should fall back to the default behavior + /// followed when poison message handling is not enabled. + /// + /// The entity instance the event was sent to, or null + /// if this information is not available. + /// The "poisoned" history event. + /// The category describing why the event is "poisoned". + /// A human-readable description of why the event is "poisoned". + /// True if the poison message was successfully handled, otherwise false. + public Task HandlePoisonEntityMessageAsync(OrchestrationInstance? entityInstance, HistoryEvent historyEvent, PoisonMessageReason reason, string details); + + /// + /// Invoked to handle a work item that is invalid and cannot be processed at all. + /// + /// + /// If this method returns false, the dispatcher should fall back to the default behavior + /// followed in the case of an invalid work item. + /// + /// The work item that could not be processed. + /// The category describing why the work item is invalid. + /// A human-readable description of why the work item is invalid. + /// Indicates whether the work item is for an entity. + /// True if the poison message was successfully handled, otherwise false. + public Task HandleInvalidWorkItemAsync(TaskOrchestrationWorkItem workItem, PoisonMessageReason reason, string details, bool isEntity); + + /// + /// Invoked to handle a work item that is invalid and cannot be processed at all. + /// + /// + /// If this method returns false, the dispatcher should fall back to the default behavior + /// followed in the case of an invalid work item. + /// + /// The work item that could not be processed. + /// The category describing why the work item is invalid. + /// A human-readable description of why the work item is invalid. + /// True if the poison message was successfully handled, otherwise false. + public Task HandleInvalidWorkItemAsync(TaskActivityWorkItem workItem, PoisonMessageReason reason, string details); + } +} diff --git a/src/DurableTask.Core/Logging/EventIds.cs b/src/DurableTask.Core/Logging/EventIds.cs index 049af43f3..f8cb6d769 100644 --- a/src/DurableTask.Core/Logging/EventIds.cs +++ b/src/DurableTask.Core/Logging/EventIds.cs @@ -72,5 +72,6 @@ static class EventIds public const int OrchestrationDebugTrace = 73; public const int OrchestrationCompletedWithWarning = 74; + public const int PoisonMessageDetected = 75; } } diff --git a/src/DurableTask.Core/Logging/LogEvents.cs b/src/DurableTask.Core/Logging/LogEvents.cs index e5fe2ee33..36f6c8d10 100644 --- a/src/DurableTask.Core/Logging/LogEvents.cs +++ b/src/DurableTask.Core/Logging/LogEvents.cs @@ -1976,5 +1976,59 @@ void IEventSourceEvent.WriteEventSource() => Utils.PackageVersion); } + /// + /// Log event representing the discarding of a "poison" message. + /// + internal class PoisonMessageDetected : StructuredLogEvent, IEventSourceEvent + { + public PoisonMessageDetected(OrchestrationInstance orchestrationInstance, string eventType, int taskEventId, int dispatchCount, string details) + { + this.InstanceId = orchestrationInstance?.InstanceId ?? string.Empty; + this.ExecutionId = orchestrationInstance?.ExecutionId ?? string.Empty; + this.EventType = eventType; + this.TaskEventId = taskEventId; + this.Details = details; + this.DispatchCount = dispatchCount; + } + + [StructuredLogField] + public string InstanceId { get; } + + [StructuredLogField] + public string ExecutionId { get; } + + [StructuredLogField] + public string EventType { get; } + + [StructuredLogField] + public int TaskEventId { get; } + + [StructuredLogField] + public int DispatchCount { get; } + + [StructuredLogField] + public string Details { get; } + + public override EventId EventId => new EventId( + EventIds.PoisonMessageDetected, + nameof(EventIds.PoisonMessageDetected)); + + public override LogLevel Level => LogLevel.Warning; + + protected override string CreateLogMessage() => + $"{this.InstanceId}: Poison message detected for {GetEventDescription(this.EventType, this.TaskEventId)} with dispatch count {this.DispatchCount}: {this.Details}"; + + void IEventSourceEvent.WriteEventSource() => + StructuredEventSource.Log.PoisonMessageDetected( + this.InstanceId, + this.ExecutionId, + this.EventType, + this.TaskEventId, + this.DispatchCount, + this.Details, + Utils.AppName, + Utils.PackageVersion); + } + } } diff --git a/src/DurableTask.Core/Logging/LogHelper.cs b/src/DurableTask.Core/Logging/LogHelper.cs index 6efbe0cfe..37216df13 100644 --- a/src/DurableTask.Core/Logging/LogHelper.cs +++ b/src/DurableTask.Core/Logging/LogHelper.cs @@ -17,6 +17,8 @@ namespace DurableTask.Core.Logging using System.Collections.Generic; using System.Text; using DurableTask.Core.Command; + using DurableTask.Core.Common; + using DurableTask.Core.Entities.EventFormat; using DurableTask.Core.Entities.OperationFormat; using DurableTask.Core.History; using Microsoft.Extensions.Logging; @@ -760,6 +762,64 @@ internal void RenewActivityMessageFailed(TaskActivityWorkItem workItem, Exceptio this.WriteStructuredLog(new LogEvents.RenewActivityMessageFailed(workItem, exception), exception); } } + + /// + /// Logs that a "poison message" has been detected and is being dropped. + /// + /// The orchestration instance this event was sent to. + /// The "poisoned" event. + /// Extra details related to the processing of this poison message. + internal void PoisonMessageDetected(OrchestrationInstance? orchestrationInstance, HistoryEvent historyEvent, string details) + { + if (this.IsStructuredLoggingEnabled) + { + this.WriteStructuredLog(new LogEvents.PoisonMessageDetected( + orchestrationInstance, + historyEvent.EventType.ToString(), + Utils.GetTaskEventId(historyEvent), + historyEvent.DispatchCount, + details)); + } + } + + /// + /// Logs that a "poison" entity request message has been detected and is being dropped. + /// + /// The orchestration instance this event was sent to. + /// The "poisoned" request message. + /// Extra details related to the processing of this poison message. + internal void PoisonMessageDetected(OrchestrationInstance orchestrationInstance, RequestMessage requestMessage, string details) + { + if (this.IsStructuredLoggingEnabled) + { + this.WriteStructuredLog(new LogEvents.PoisonMessageDetected( + orchestrationInstance, + requestMessage.IsLockRequest ? "LockRequest" : "OperationRequest", + taskEventId: -1, + requestMessage.DispatchCount, + details)); + } + } + + /// + /// Logs that a "poison" entity lock release message has been detected and is being dropped. + /// + /// The orchestration instance this event was sent to. + /// The "poisoned" release message. + /// The dispatch count of the release message. + /// Extra details related to the processing of this poison message. + internal void PoisonMessageDetected(OrchestrationInstance orchestrationInstance, ReleaseMessage releaseMessage, int dispatchCount, string details) + { + if (this.IsStructuredLoggingEnabled) + { + this.WriteStructuredLog(new LogEvents.PoisonMessageDetected( + orchestrationInstance, + "LockRelease", + taskEventId: -1, + dispatchCount, + details)); + } + } #endregion internal void OrchestrationDebugTrace(string instanceId, string executionId, string details) diff --git a/src/DurableTask.Core/Logging/StructuredEventSource.cs b/src/DurableTask.Core/Logging/StructuredEventSource.cs index 129bac158..fb09dc905 100644 --- a/src/DurableTask.Core/Logging/StructuredEventSource.cs +++ b/src/DurableTask.Core/Logging/StructuredEventSource.cs @@ -656,6 +656,33 @@ internal void DiscardingMessage( } } + [Event(EventIds.PoisonMessageDetected, Level = EventLevel.Warning, Version = 1)] + internal void PoisonMessageDetected( + string InstanceId, + string ExecutionId, + string EventType, + int TaskEventId, + int DispatchCount, + string Details, + string AppName, + string ExtensionVersion) + { + if (this.IsEnabled(EventLevel.Warning)) + { + // TODO: Use WriteEventCore for better performance + this.WriteEvent( + EventIds.PoisonMessageDetected, + InstanceId ?? string.Empty, + ExecutionId ?? string.Empty, + EventType, + TaskEventId, + DispatchCount, + Details, + AppName, + ExtensionVersion); + } + } + [Event(EventIds.EntityBatchExecuting, Level = EventLevel.Informational, Version = 1)] internal void EntityBatchExecuting( string InstanceId, diff --git a/src/DurableTask.Core/TaskActivityDispatcher.cs b/src/DurableTask.Core/TaskActivityDispatcher.cs index 8f4c24dca..85ba81b48 100644 --- a/src/DurableTask.Core/TaskActivityDispatcher.cs +++ b/src/DurableTask.Core/TaskActivityDispatcher.cs @@ -37,6 +37,7 @@ public sealed class TaskActivityDispatcher readonly LogHelper logHelper; readonly ErrorPropagationMode errorPropagationMode; readonly IExceptionPropertiesProvider? exceptionPropertiesProvider; + readonly IPoisonMessageHandler? poisonMessageHandler; /// /// Initializes a new instance of the class with an exception properties provider. @@ -61,6 +62,7 @@ internal TaskActivityDispatcher( this.logHelper = logHelper; this.errorPropagationMode = errorPropagationMode; this.exceptionPropertiesProvider = exceptionPropertiesProvider; + this.poisonMessageHandler = orchestrationService as IPoisonMessageHandler; this.dispatcher = new WorkItemDispatcher( "TaskActivityDispatcher", @@ -117,9 +119,14 @@ async Task OnProcessWorkItemAsync(TaskActivityWorkItem workItem) { if (orchestrationInstance == null || string.IsNullOrWhiteSpace(orchestrationInstance.InstanceId)) { - this.logHelper.TaskActivityDispatcherError( - workItem, - $"The activity worker received a message that does not have any OrchestrationInstance information."); + string message = "The activity worker received a message that does not have any OrchestrationInstance information."; + if (this.poisonMessageHandler != null + && await this.poisonMessageHandler.HandleInvalidWorkItemAsync(workItem, PoisonMessageReason.MissingOrchestrationInstanceId, message)) + { + this.logHelper.PoisonMessageDetected(orchestrationInstance, taskMessage.Event, message); + return; + } + this.logHelper.TaskActivityDispatcherError(workItem, message); throw TraceHelper.TraceException( TraceEventType.Error, "TaskActivityDispatcher-MissingOrchestrationInstance", @@ -128,14 +135,20 @@ async Task OnProcessWorkItemAsync(TaskActivityWorkItem workItem) if (taskMessage.Event.EventType != EventType.TaskScheduled) { - this.logHelper.TaskActivityDispatcherError( - workItem, - $"The activity worker received an event of type '{taskMessage.Event.EventType}' but only '{EventType.TaskScheduled}' is supported."); + string message = $"The activity worker received an event of type '{taskMessage.Event.EventType}' but only " + + $"'{EventType.TaskScheduled}' is supported."; + if (this.poisonMessageHandler != null + && await this.poisonMessageHandler.HandleInvalidWorkItemAsync(workItem, PoisonMessageReason.WrongEventType, message)) + { + this.logHelper.PoisonMessageDetected(orchestrationInstance, taskMessage.Event, message); + return; + } + this.logHelper.TaskActivityDispatcherError(workItem, message); throw TraceHelper.TraceException( TraceEventType.Critical, "TaskActivityDispatcher-UnsupportedEventType", new NotSupportedException("Activity worker does not support event of type: " + - taskMessage.Event.EventType)); + taskMessage.Event.EventType)); } scheduledEvent = (TaskScheduledEvent)taskMessage.Event; @@ -143,136 +156,178 @@ async Task OnProcessWorkItemAsync(TaskActivityWorkItem workItem) // Distributed tracing: start a new trace activity derived from the orchestration's trace context. Activity? traceActivity = TraceHelper.StartTraceActivityForTaskExecution(scheduledEvent, orchestrationInstance); + string? poisonMessageReason = null; if (scheduledEvent.Name == null) { - string message = $"The activity worker received a {nameof(EventType.TaskScheduled)} event that does not specify an activity name."; - this.logHelper.TaskActivityDispatcherError(workItem, message); - throw TraceHelper.TraceException( - TraceEventType.Error, - "TaskActivityDispatcher-MissingActivityName", - new InvalidOperationException(message)); + poisonMessageReason = $"The activity worker received a {nameof(EventType.TaskScheduled)} event that does not specify an activity name."; + if (this.poisonMessageHandler != null + && await this.poisonMessageHandler.HandleInvalidWorkItemAsync(workItem, PoisonMessageReason.MissingActivityName, poisonMessageReason)) + { + this.logHelper.PoisonMessageDetected(orchestrationInstance, taskMessage.Event, poisonMessageReason); + // We want to keep processing in this case to fail the calling orchestration, so we don't return here + } + else + { + this.logHelper.TaskActivityDispatcherError(workItem, poisonMessageReason); + throw TraceHelper.TraceException( + TraceEventType.Error, + "TaskActivityDispatcher-MissingActivityName", + new InvalidOperationException(poisonMessageReason)); + } } - this.logHelper.TaskActivityStarting(orchestrationInstance, scheduledEvent); - TaskActivity? taskActivity = this.objectManager.GetObject(scheduledEvent.Name, scheduledEvent.Version); - - if (workItem.LockedUntilUtc < DateTime.MaxValue) + if (poisonMessageReason == null && scheduledEvent.DispatchCount > this.poisonMessageHandler?.MaxDispatchCount) { - // start a task to run RenewUntil - renewTask = Task.Factory.StartNew( - () => this.RenewUntil(workItem, renewCancellationTokenSource.Token), - renewCancellationTokenSource.Token); + poisonMessageReason = $"Activity has received an event with dispatch count {taskMessage.Event.DispatchCount} which exceeds the maximum dispatch " + + $"count of {this.poisonMessageHandler.MaxDispatchCount}. The task will be failed."; } - var dispatchContext = new DispatchMiddlewareContext(); - dispatchContext.SetProperty(taskMessage.OrchestrationInstance); - dispatchContext.SetProperty(taskActivity); - dispatchContext.SetProperty(scheduledEvent); - - // In transitionary phase (activity queued from old code, accessed in new code) context can be null. - if (taskMessage.OrchestrationExecutionContext != null) + HistoryEvent? eventToRespond = null; + if (poisonMessageReason != null) { - dispatchContext.SetProperty(taskMessage.OrchestrationExecutionContext); + this.logHelper.PoisonMessageDetected( + orchestrationInstance, + taskMessage.Event, + poisonMessageReason); + + eventToRespond = new TaskFailedEvent( + -1, + scheduledEvent.EventId, + reason: poisonMessageReason, + details: null, + new + ( + "PoisonMessage", + poisonMessageReason!, + stackTrace: null, + innerFailure: null, + isNonRetriable: true) + ); + traceActivity?.SetStatus(ActivityStatusCode.Error, poisonMessageReason); } - - // correlation - CorrelationTraceClient.Propagate(() => + else { - workItem.TraceContextBase?.SetActivityToCurrent(); - diagnosticActivity = workItem.TraceContextBase?.CurrentActivity; - }); + this.logHelper.TaskActivityStarting(orchestrationInstance, scheduledEvent); + TaskActivity? taskActivity = this.objectManager.GetObject(scheduledEvent.Name!, scheduledEvent.Version); - ActivityExecutionResult? result; - try - { - await this.dispatchPipeline.RunAsync(dispatchContext, async _ => + if (workItem.LockedUntilUtc < DateTime.MaxValue) { - if (taskActivity == null) - { - // This likely indicates a deployment error of some kind. Because these unhandled exceptions are - // automatically retried, resolving this may require redeploying the app code so that the activity exists again. - // CONSIDER: Should this be changed into a permanent error that fails the orchestration? Perhaps - // the app owner doesn't care to preserve existing instances when doing code deployments? - throw new TypeMissingException($"TaskActivity {scheduledEvent.Name} version {scheduledEvent.Version} was not found"); - } - - var context = new TaskContext( - taskMessage.OrchestrationInstance, - scheduledEvent.Name, - scheduledEvent.Version, - scheduledEvent.EventId); - context.ErrorPropagationMode = this.errorPropagationMode; - context.ExceptionPropertiesProvider = this.exceptionPropertiesProvider; - - HistoryEvent? responseEvent; - - try - { - string? output = await taskActivity.RunAsync(context, scheduledEvent.Input); - responseEvent = new TaskCompletedEvent(-1, scheduledEvent.EventId, output); - } - catch (Exception e) when (e is not TaskFailureException && !Utils.IsFatal(e) && !Utils.IsExecutionAborting(e)) - { - // These are unexpected exceptions that occur in the task activity abstraction. Normal exceptions from - // activities are expected to be translated into TaskFailureException and handled outside the middleware - // context (see further below). - TraceHelper.TraceExceptionInstance(TraceEventType.Error, "TaskActivityDispatcher-ProcessException", taskMessage.OrchestrationInstance, e); - string? details = this.IncludeDetails - ? $"Unhandled exception while executing task: {e}" - : null; - responseEvent = new TaskFailedEvent(-1, scheduledEvent.EventId, e.Message, details, new FailureDetails(e)); - - traceActivity?.SetStatus(ActivityStatusCode.Error, e.Message); - - this.logHelper.TaskActivityFailure(orchestrationInstance, scheduledEvent.Name, (TaskFailedEvent)responseEvent, e); - } - - var result = new ActivityExecutionResult { ResponseEvent = responseEvent }; - dispatchContext.SetProperty(result); + // start a task to run RenewUntil + renewTask = Task.Factory.StartNew( + () => this.RenewUntil(workItem, renewCancellationTokenSource.Token), + renewCancellationTokenSource.Token); + } + + var dispatchContext = new DispatchMiddlewareContext(); + dispatchContext.SetProperty(taskMessage.OrchestrationInstance); + dispatchContext.SetProperty(taskActivity); + dispatchContext.SetProperty(scheduledEvent); + + // In transitionary phase (activity queued from old code, accessed in new code) context can be null. + if (taskMessage.OrchestrationExecutionContext != null) + { + dispatchContext.SetProperty(taskMessage.OrchestrationExecutionContext); + } + + // correlation + CorrelationTraceClient.Propagate(() => + { + workItem.TraceContextBase?.SetActivityToCurrent(); + diagnosticActivity = workItem.TraceContextBase?.CurrentActivity; }); - result = dispatchContext.GetProperty(); - } - catch (TaskFailureException e) - { - // These are normal task activity failures. They can come from Activity implementations or from middleware. - TraceHelper.TraceExceptionInstance(TraceEventType.Error, "TaskActivityDispatcher-ProcessTaskFailure", taskMessage.OrchestrationInstance, e); - string? details = this.IncludeDetails ? e.Details : null; - var failureEvent = new TaskFailedEvent(-1, scheduledEvent.EventId, e.Message, details, e.FailureDetails); + ActivityExecutionResult? result; + try + { + await this.dispatchPipeline.RunAsync(dispatchContext, async _ => + { + if (taskActivity == null) + { + // This likely indicates a deployment error of some kind. Because these unhandled exceptions are + // automatically retried, resolving this may require redeploying the app code so that the activity exists again. + // CONSIDER: Should this be changed into a permanent error that fails the orchestration? Perhaps + // the app owner doesn't care to preserve existing instances when doing code deployments? + throw new TypeMissingException($"TaskActivity {scheduledEvent.Name} version {scheduledEvent.Version} was not found"); + } + + var context = new TaskContext( + taskMessage.OrchestrationInstance, + scheduledEvent.Name!, + scheduledEvent.Version, + scheduledEvent.EventId); + context.ErrorPropagationMode = this.errorPropagationMode; + context.ExceptionPropertiesProvider = this.exceptionPropertiesProvider; + + HistoryEvent? responseEvent; + + try + { + string? output = await taskActivity.RunAsync(context, scheduledEvent.Input); + responseEvent = new TaskCompletedEvent(-1, scheduledEvent.EventId, output); + } + catch (Exception e) when (e is not TaskFailureException && !Utils.IsFatal(e) && !Utils.IsExecutionAborting(e)) + { + // These are unexpected exceptions that occur in the task activity abstraction. Normal exceptions from + // activities are expected to be translated into TaskFailureException and handled outside the middleware + // context (see further below). + TraceHelper.TraceExceptionInstance(TraceEventType.Error, "TaskActivityDispatcher-ProcessException", taskMessage.OrchestrationInstance, e); + string? details = this.IncludeDetails + ? $"Unhandled exception while executing task: {e}" + : null; + responseEvent = new TaskFailedEvent(-1, scheduledEvent.EventId, e.Message, details, new FailureDetails(e)); + + traceActivity?.SetStatus(ActivityStatusCode.Error, e.Message); + + this.logHelper.TaskActivityFailure(orchestrationInstance, scheduledEvent.Name!, (TaskFailedEvent)responseEvent, e); + } + + var result = new ActivityExecutionResult { ResponseEvent = responseEvent }; + dispatchContext.SetProperty(result); + }); + + result = dispatchContext.GetProperty(); + } + catch (TaskFailureException e) + { + // These are normal task activity failures. They can come from Activity implementations or from middleware. + TraceHelper.TraceExceptionInstance(TraceEventType.Error, "TaskActivityDispatcher-ProcessTaskFailure", taskMessage.OrchestrationInstance, e); + string? details = this.IncludeDetails ? e.Details : null; + var failureEvent = new TaskFailedEvent(-1, scheduledEvent.EventId, e.Message, details, e.FailureDetails); - traceActivity?.SetStatus(ActivityStatusCode.Error, e.Message); + traceActivity?.SetStatus(ActivityStatusCode.Error, e.Message); - this.logHelper.TaskActivityFailure(orchestrationInstance, scheduledEvent.Name, failureEvent, e); - CorrelationTraceClient.Propagate(() => CorrelationTraceClient.TrackException(e)); - result = new ActivityExecutionResult { ResponseEvent = failureEvent }; - } - catch (Exception middlewareException) when (!Utils.IsFatal(middlewareException)) - { - traceActivity?.SetStatus(ActivityStatusCode.Error, middlewareException.Message); + this.logHelper.TaskActivityFailure(orchestrationInstance, scheduledEvent.Name!, failureEvent, e); + CorrelationTraceClient.Propagate(() => CorrelationTraceClient.TrackException(e)); + result = new ActivityExecutionResult { ResponseEvent = failureEvent }; + } + catch (Exception middlewareException) when (!Utils.IsFatal(middlewareException)) + { + traceActivity?.SetStatus(ActivityStatusCode.Error, middlewareException.Message); - // These are considered retriable - this.logHelper.TaskActivityDispatcherError(workItem, $"Unhandled exception in activity middleware pipeline: {middlewareException}"); - throw; - } + // These are considered retriable + this.logHelper.TaskActivityDispatcherError(workItem, $"Unhandled exception in activity middleware pipeline: {middlewareException}"); + throw; + } - HistoryEvent? eventToRespond = result?.ResponseEvent; + eventToRespond = result?.ResponseEvent; - if (eventToRespond is TaskCompletedEvent completedEvent) - { - this.logHelper.TaskActivityCompleted(orchestrationInstance, scheduledEvent.Name, completedEvent); - } - else if (eventToRespond is null) - { - // Default response if middleware prevents a response from being generated - eventToRespond = new TaskCompletedEvent(-1, scheduledEvent.EventId, null); - } + if (eventToRespond is TaskCompletedEvent completedEvent) + { + this.logHelper.TaskActivityCompleted(orchestrationInstance, scheduledEvent.Name!, completedEvent); + } + else if (eventToRespond is null) + { + // Default response if middleware prevents a response from being generated + eventToRespond = new TaskCompletedEvent(-1, scheduledEvent.EventId, null); + } - if (traceActivity != null && eventToRespond is TaskCompletedEvent) - { - // Ensure successful executions don't preserve a prior error status from custom instrumentation. - traceActivity.SetStatus(ActivityStatusCode.OK, "Completed"); + if (traceActivity != null && eventToRespond is TaskCompletedEvent) + { + // Ensure successful executions don't preserve a prior error status from custom instrumentation. + traceActivity.SetStatus(ActivityStatusCode.OK, "Completed"); + } } - + var responseTaskMessage = new TaskMessage { Event = eventToRespond, diff --git a/src/DurableTask.Core/TaskEntityDispatcher.cs b/src/DurableTask.Core/TaskEntityDispatcher.cs index a3d163e28..216b98233 100644 --- a/src/DurableTask.Core/TaskEntityDispatcher.cs +++ b/src/DurableTask.Core/TaskEntityDispatcher.cs @@ -1,4 +1,4 @@ -// ---------------------------------------------------------------------------------- +// ---------------------------------------------------------------------------------- // 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. @@ -25,8 +25,10 @@ namespace DurableTask.Core using System; using System.Collections.Generic; using System.Diagnostics; + using System.Linq; using System.Threading; using System.Threading.Tasks; + using static DurableTask.Core.TaskOrchestrationDispatcher; /// /// Dispatcher for orchestrations and entities to handle processing and renewing, completion of orchestration events. @@ -43,6 +45,7 @@ public class TaskEntityDispatcher readonly ErrorPropagationMode errorPropagationMode; readonly TaskOrchestrationDispatcher.NonBlockingCountdownLock concurrentSessionLock; readonly IExceptionPropertiesProvider exceptionPropertiesProvider; + readonly IPoisonMessageHandler poisonMessageHandler; /// /// Initializes a new instance of the class with an exception properties provider. @@ -69,7 +72,8 @@ internal TaskEntityDispatcher( this.exceptionPropertiesProvider = exceptionPropertiesProvider; this.entityOrchestrationService = (orchestrationService as IEntityOrchestrationService)!; this.entityBackendProperties = entityOrchestrationService.EntityBackendProperties; - + this.poisonMessageHandler = orchestrationService as IPoisonMessageHandler; + this.dispatcher = new WorkItemDispatcher( "TaskEntityDispatcher", item => item == null ? string.Empty : item.InstanceId, @@ -155,8 +159,8 @@ async Task OnProcessWorkItemSessionAsync(TaskOrchestrationWorkItem workItem) // If we failed to acquire it, we will end the extended session after this execution. schedulerState = await this.OnProcessWorkItemAsync(workItem, schedulerState); - // The entity has been deleted, so we end the extended session. - if (this.EntityIsDeleted(schedulerState)) + // The work item could not be processed or the entity has been deleted, so we end the extended session. + if (schedulerState == null || this.EntityIsDeleted(schedulerState)) { break; } @@ -251,23 +255,53 @@ private async Task OnProcessWorkItemAsync(TaskOrchestrationWorkI try { - // Assumes that: if the batch contains a new "ExecutionStarted" event, it is the first message in the batch. - if (!TaskOrchestrationDispatcher.ReconcileMessagesWithState(workItem, nameof(TaskEntityDispatcher), this.errorPropagationMode, this.logHelper)) + var reconcileResult = await ReconcileMessagesWithStateAsync( + workItem, + nameof(TaskEntityDispatcher), + this.errorPropagationMode, + this.logHelper, + this.poisonMessageHandler, + isEntity: true); + + if (reconcileResult != WorkItemReconcileResult.Success) { // TODO : mark an orchestration as faulted if there is data corruption this.logHelper.DroppingOrchestrationWorkItem(workItem, "Received work-item for an invalid orchestration"); + + if (reconcileResult == WorkItemReconcileResult.PoisonMessageHandled) + { + // Signal the extended session to end if one is running + return null; + } } else { bool firstExecutionIfExtendedSession = schedulerState == null; + Work workToDoNow = null; + // we start with processing all the requests and figuring out which ones to execute now // results can depend on whether the entity is locked, what the maximum batch size is, // and whether the messages arrived out of order + DetermineWorkResult determineWorkResult = await this.DetermineWorkAsync(workItem, schedulerState); + schedulerState = determineWorkResult.SchedulerState; + workToDoNow = determineWorkResult.Batch; - this.DetermineWork(workItem.OrchestrationRuntimeState, - ref schedulerState, - out Work workToDoNow); + if (!determineWorkResult.Success) + { + this.logHelper.DroppingOrchestrationWorkItem(workItem, determineWorkResult.ErrorMessage); + + if (this.poisonMessageHandler != null + && await this.poisonMessageHandler.HandleInvalidWorkItemAsync( + workItem, + PoisonMessageReason.DeserializationError, + determineWorkResult.ErrorMessage, + isEntity: true)) + { + // Signal the extended session to end if one is running + return null; + } + } if (workToDoNow.OperationCount > 0) { @@ -437,14 +471,18 @@ await this.orchestrationService.CompleteTaskOrchestrationWorkItemAsync( void ProcessLockRequest(WorkItemEffects effects, SchedulerState schedulerState, RequestMessage request) { - this.logHelper.EntityLockAcquired(effects.InstanceId, request); + bool isPoisonMessage = request.DispatchCount > this.poisonMessageHandler?.MaxDispatchCount; + if (!isPoisonMessage) + { + this.logHelper.EntityLockAcquired(effects.InstanceId, request); - // mark the entity state as locked - schedulerState.LockedBy = request.ParentInstanceId; + // mark the entity state as locked + schedulerState.LockedBy = request.ParentInstanceId; - request.Position++; + request.Position++; + } - if (request.Position < request.LockSet.Length) + if (request.Position < request.LockSet.Length && !isPoisonMessage) { // send lock request to next entity in the lock set var target = new OrchestrationInstance() { InstanceId = request.LockSet[request.Position].ToString() }; @@ -454,7 +492,22 @@ void ProcessLockRequest(WorkItemEffects effects, SchedulerState schedulerState, { // send lock acquisition completed response back to originating orchestration instance var target = new OrchestrationInstance() { InstanceId = request.ParentInstanceId, ExecutionId = request.ParentExecutionId }; - this.SendLockResponseMessage(effects, target, request.Id); + + // In the case of a poison message, it will be the locking instance's responsibility to unlock any other entities for whom the + // lock request may have succeeded. + this.SendLockResponseMessage( + effects, + target, + request.Id, + isPoisonMessage ? + new FailureDetails( + "PoisonMessage", + $"Entity lock request has dispatch count {request.DispatchCount} " + + $"which exceeds the maximum dispatch count of {this.poisonMessageHandler.MaxDispatchCount}.", + stackTrace: null, + innerFailure: null, + isNonRetriable: true) + : null); } } @@ -475,12 +528,29 @@ string SerializeSchedulerStateForNextExecution(SchedulerState schedulerState) #region Preprocess to determine work - void DetermineWork(OrchestrationRuntimeState runtimeState, ref SchedulerState schedulerState, out Work batch) + readonly struct DetermineWorkResult + { + public DetermineWorkResult(bool success, SchedulerState schedulerState, Work batch, string errorMessage) + { + this.Success = success; + this.SchedulerState = schedulerState; + this.Batch = batch; + this.ErrorMessage = errorMessage; + } + + public bool Success { get; } + public SchedulerState SchedulerState { get; } + public Work Batch { get; } + public string ErrorMessage { get; } + } + + async Task DetermineWorkAsync(TaskOrchestrationWorkItem workItem, SchedulerState schedulerState) { + OrchestrationRuntimeState runtimeState = workItem.OrchestrationRuntimeState; string instanceId = runtimeState.OrchestrationInstance.InstanceId; bool deserializeState = schedulerState == null; schedulerState ??= new(); - batch = new Work(); + Work batch = new Work(); Queue lockHolderMessages = null; @@ -501,6 +571,15 @@ void DetermineWork(OrchestrationRuntimeState runtimeState, ref SchedulerState sc } catch (Exception exception) { + if (this.poisonMessageHandler != null) + { + string errorMessage = $"Failed to deserialize the entity scheduler state from the {EventType.ExecutionStarted} input."; + this.logHelper.PoisonMessageDetected( + runtimeState.OrchestrationInstance, + e, + $"Dropping entity work item: {errorMessage}"); + return new DetermineWorkResult(success: false, schedulerState, batch, errorMessage); + } throw new EntitySchedulerException("Failed to deserialize entity scheduler state - may be corrupted or wrong version.", exception); } } @@ -508,6 +587,7 @@ void DetermineWork(OrchestrationRuntimeState runtimeState, ref SchedulerState sc case EventType.EventRaised: EventRaisedEvent eventRaisedEvent = (EventRaisedEvent)e; + bool isPoisonMessage = eventRaisedEvent.DispatchCount > this.poisonMessageHandler?.MaxDispatchCount; if (EntityMessageEventNames.IsRequestMessage(eventRaisedEvent.Name)) { @@ -520,9 +600,26 @@ void DetermineWork(OrchestrationRuntimeState runtimeState, ref SchedulerState sc } catch (Exception exception) { - throw new EntitySchedulerException("Failed to deserialize incoming request message - may be corrupted or wrong version.", exception); + string failureReason = $"Failed to deserialize incoming entity request message - may be corrupted or wrong version: {exception.Message}"; + if (this.poisonMessageHandler != null) + { + this.logHelper.PoisonMessageDetected( + runtimeState.OrchestrationInstance, + eventRaisedEvent, + failureReason); + if (await this.poisonMessageHandler.HandlePoisonEntityMessageAsync( + workItem.OrchestrationRuntimeState.OrchestrationInstance, + eventRaisedEvent, + PoisonMessageReason.DeserializationError, + failureReason)) + { + break; + } + } + throw new EntitySchedulerException(failureReason, exception); } + requestMessage.DispatchCount = eventRaisedEvent.DispatchCount; IEnumerable deliverNow; if (requestMessage.ScheduledTime.HasValue) @@ -585,7 +682,38 @@ void DetermineWork(OrchestrationRuntimeState runtimeState, ref SchedulerState sc } catch (Exception exception) { - throw new EntitySchedulerException("Failed to deserialize lock release message - may be corrupted or wrong version.", exception); + string failureReason = $"Failed to deserialize entity lock release message - may be corrupted or wrong version: {exception.Message}"; + if (this.poisonMessageHandler != null) + { + this.logHelper.PoisonMessageDetected( + runtimeState.OrchestrationInstance, + eventRaisedEvent, + failureReason); + if (await this.poisonMessageHandler.HandlePoisonEntityMessageAsync( + workItem.OrchestrationRuntimeState.OrchestrationInstance, + eventRaisedEvent, + PoisonMessageReason.DeserializationError, + failureReason)) + { + break; + } + } + throw new EntitySchedulerException(failureReason, exception); + } + + if (isPoisonMessage) + { + string failureReason = $"Entity lock release message from parent instance '{message.ParentInstanceId}' has dispatch count " + + $"{eventRaisedEvent.DispatchCount} which exceeds the maximum allowed dispatch count of {this.poisonMessageHandler.MaxDispatchCount}."; + this.logHelper.PoisonMessageDetected(runtimeState.OrchestrationInstance, message, eventRaisedEvent.DispatchCount, failureReason); + if (await this.poisonMessageHandler.HandlePoisonEntityMessageAsync( + workItem.OrchestrationRuntimeState.OrchestrationInstance, + eventRaisedEvent, + PoisonMessageReason.DispatchCount, + failureReason)) + { + break; + } } if (schedulerState.LockedBy == message.ParentInstanceId) @@ -596,6 +724,21 @@ void DetermineWork(OrchestrationRuntimeState runtimeState, ref SchedulerState sc } else { + if (isPoisonMessage) + { + string failureReason = $"Entity self-continue message has dispatch count {eventRaisedEvent.DispatchCount} which exceeds the maximum allowed " + + $"dispatch count of {this.poisonMessageHandler.MaxDispatchCount}."; + this.logHelper.PoisonMessageDetected(runtimeState.OrchestrationInstance, eventRaisedEvent, failureReason); + if (await this.poisonMessageHandler.HandlePoisonEntityMessageAsync( + workItem.OrchestrationRuntimeState.OrchestrationInstance, + eventRaisedEvent, + PoisonMessageReason.DispatchCount, + failureReason)) + { + break; + } + } + // this is a continue message. // Resumes processing of previously queued operations, if any. schedulerState.Suspended = false; @@ -627,6 +770,15 @@ void DetermineWork(OrchestrationRuntimeState runtimeState, ref SchedulerState sc } var request = schedulerState.Dequeue(); + var poisonMessageHandler = this.poisonMessageHandler; + if (poisonMessageHandler != null && request.DispatchCount > poisonMessageHandler.MaxDispatchCount) + { + this.logHelper.PoisonMessageDetected( + runtimeState.OrchestrationInstance, + request, + $"Entity request has dispatch count {request.DispatchCount} which exceeds the maximum dispatch count " + + $"of {poisonMessageHandler.MaxDispatchCount} and will be failed."); + } if (request.IsLockRequest) { @@ -639,6 +791,8 @@ void DetermineWork(OrchestrationRuntimeState runtimeState, ref SchedulerState sc } } } + + return new DetermineWorkResult(success: true, schedulerState, batch, errorMessage: null); } bool EntityIsDeleted(SchedulerState schedulerState) @@ -721,7 +875,7 @@ public void ToBeContinued(SchedulerState schedulerState) parentTraceContext); } - // We still want to add the trace activity to the list even if it was not successfully created and is null. This is because otherwise we have no easy way of mapping OperationResults to Activities otherwise if the lists + // We still want to add the trace activity to the list even if it was not successfully created and is null. This is because otherwise we have no easy way of mapping OperationResults to Activities if the lists // do not have the same length in TraceHelper.EndActivitiesForProcessingEntityInvocation. We will simply skip ending the Activity if it is null in this method traceActivities.Add(traceActivity); @@ -846,12 +1000,13 @@ void SendLockRequestMessage(WorkItemEffects effects, SchedulerState schedulerSta this.ProcessSendEventMessage(effects, target, EntityMessageEventNames.RequestMessageEventName, message); } - void SendLockResponseMessage(WorkItemEffects effects, OrchestrationInstance target, Guid requestId) + void SendLockResponseMessage(WorkItemEffects effects, OrchestrationInstance target, Guid requestId, FailureDetails failureDetails) { var message = new ResponseMessage() { // content is ignored by receiver but helps with tracing - Result = ResponseMessage.LockAcquisitionCompletion, + Result = ResponseMessage.LockAcquisitionCompletion, + FailureDetails = failureDetails, }; this.ProcessSendEventMessage(effects, target, EntityMessageEventNames.ResponseMessageEventName(requestId), message); } @@ -955,13 +1110,30 @@ internal void ProcessSendStartMessage(WorkItemEffects effects, OrchestrationRunt async Task ExecuteViaMiddlewareAsync(Work workToDoNow, OrchestrationInstance instance, string serializedEntityState, bool isExtendedSession, bool includeEntityState) { + var startTime = DateTime.UtcNow; var (operations, traceActivities) = workToDoNow.GetOperationRequestsAndTraceActivities(instance.InstanceId); + + bool poisonMessagesExist = workToDoNow.Operations.Any(op => op.DispatchCount > this.poisonMessageHandler?.MaxDispatchCount); + var operationsToSend = operations; + + if (poisonMessagesExist) + { + operationsToSend = new List(); + for (int i = 0; i < operations.Count; i++) + { + if (workToDoNow.Operations[i].DispatchCount <= this.poisonMessageHandler.MaxDispatchCount) + { + operationsToSend.Add(operations[i]); + } + } + } + // the request object that will be passed to the worker var request = new EntityBatchRequest() { InstanceId = instance.InstanceId, EntityState = serializedEntityState, - Operations = operations, + Operations = operationsToSend, }; this.logHelper.EntityBatchExecuting(request); @@ -999,11 +1171,50 @@ await this.dispatchPipeline.RunAsync(dispatchContext, async _ => } var result = await taskEntity.ExecuteOperationBatchAsync(request); - + dispatchContext.SetProperty(result); }); var result = dispatchContext.GetProperty(); + + if (poisonMessagesExist) + { + // We initialize with an initial capacity of at least the middleware operations count, + // though we will have more results if there are poison messages + var resultAfterPoisonMessageHandling = new List(result.Results.Count); + int middlewareResultIndex = 0; + + // We end iteration once we reach the end of the middleware results, any remaining operations + // (including potential poison messages) will be deferred + for (int i = 0; i < operations.Count && middlewareResultIndex < result.Results.Count; i++) + { + if (workToDoNow.Operations[i].DispatchCount <= this.poisonMessageHandler.MaxDispatchCount) + { + resultAfterPoisonMessageHandling.Add(result.Results[middlewareResultIndex++]); + } + else + { + resultAfterPoisonMessageHandling.Add( + new() + { + FailureDetails = new + ( + "PoisonMessage", + $"Entity operation request has dispatch count {workToDoNow.Operations[i].DispatchCount} " + + $"which exceeds the maximum dispatch count of {this.poisonMessageHandler.MaxDispatchCount}.", + stackTrace: null, + innerFailure: null, + isNonRetriable: true + ), + StartTimeUtc = startTime, + EndTimeUtc = DateTime.UtcNow + } + ); + } + } + result.Results = resultAfterPoisonMessageHandling; + } + TraceHelper.EndActivitiesForProcessingEntityInvocation(traceActivities, result.Results, result.FailureDetails); this.logHelper.EntityBatchExecuted(request, result); diff --git a/src/DurableTask.Core/TaskOrchestrationDispatcher.cs b/src/DurableTask.Core/TaskOrchestrationDispatcher.cs index 751a64b78..efb1e95d6 100644 --- a/src/DurableTask.Core/TaskOrchestrationDispatcher.cs +++ b/src/DurableTask.Core/TaskOrchestrationDispatcher.cs @@ -50,6 +50,7 @@ public class TaskOrchestrationDispatcher readonly TaskOrchestrationEntityParameters? entityParameters; readonly VersioningSettings? versioningSettings; readonly IExceptionPropertiesProvider? exceptionPropertiesProvider; + readonly IPoisonMessageHandler? poisonMessageHandler; /// /// Initializes a new instance of the class with an exception properties provider. @@ -80,6 +81,7 @@ internal TaskOrchestrationDispatcher( this.entityParameters = TaskOrchestrationEntityParameters.FromEntityBackendProperties(this.entityBackendProperties); this.versioningSettings = versioningSettings; this.exceptionPropertiesProvider = exceptionPropertiesProvider; + this.poisonMessageHandler = orchestrationService as IPoisonMessageHandler; this.dispatcher = new WorkItemDispatcher( "TaskOrchestrationDispatcher", @@ -374,7 +376,14 @@ protected async Task OnProcessWorkItemAsync(TaskOrchestrationWorkItem work try { // Assumes that: if the batch contains a new "ExecutionStarted" event, it is the first message in the batch. - if (!ReconcileMessagesWithState(workItem, nameof(TaskOrchestrationDispatcher), this.errorPropagationMode, logHelper)) + var reconcileResult = await ReconcileMessagesWithStateAsync( + workItem, + nameof(TaskOrchestrationDispatcher), + this.errorPropagationMode, + logHelper, + this.poisonMessageHandler, + isEntity: this.EntitiesEnabled && Common.Entities.IsEntityInstance(workItem.InstanceId)); + if (reconcileResult != WorkItemReconcileResult.Success) { // TODO : mark an orchestration as faulted if there is data corruption this.logHelper.DroppingOrchestrationWorkItem(workItem, "Received work-item for an invalid orchestration"); @@ -385,6 +394,10 @@ protected async Task OnProcessWorkItemAsync(TaskOrchestrationWorkItem work "Received work-item for an invalid orchestration"); isCompleted = true; traceActivity?.Dispose(); + if (reconcileResult == WorkItemReconcileResult.PoisonMessageHandled) + { + return isCompleted; + } } else { @@ -433,6 +446,34 @@ protected async Task OnProcessWorkItemAsync(TaskOrchestrationWorkItem work } } + var poisonEvents = runtimeState.NewEvents.Where(evt => evt.DispatchCount > this.poisonMessageHandler?.MaxDispatchCount); + if (poisonEvents.Count() > 0) + { + foreach (var poisonEvent in poisonEvents) + { + this.logHelper.PoisonMessageDetected( + runtimeState.OrchestrationInstance!, + poisonEvent, + $"Orchestration has received an event with dispatch count {poisonEvent.DispatchCount} which exceeds the maximum dispatch " + + $"count of {this.poisonMessageHandler!.MaxDispatchCount} and will be failed."); + } + + var failureAction = new OrchestrationCompleteOrchestratorAction + { + Id = runtimeState.PastEvents.Count, + FailureDetails = new FailureDetails( + "PoisonMessages", + $"Orchestration has received messages of type {string.Join(",", poisonEvents.Select(e => e.EventType))} " + + $"with dispatch counts {string.Join(",", poisonEvents.Select(e => e.DispatchCount))} which exceed the " + + $"maximum dispatch count of {this.poisonMessageHandler!.MaxDispatchCount}.", + stackTrace: null, + innerFailure: null, + isNonRetriable: true), + OrchestrationStatus = OrchestrationStatus.Failed, + }; + decisions = new List { failureAction }; + } + this.logHelper.OrchestrationExecuting(runtimeState.OrchestrationInstance!, runtimeState.Name); TraceHelper.TraceInstance( TraceEventType.Verbose, @@ -441,7 +482,7 @@ protected async Task OnProcessWorkItemAsync(TaskOrchestrationWorkItem work "Executing user orchestration: {0}", JsonDataConverter.Default.Serialize(runtimeState.GetOrchestrationRuntimeStateDump(), true)); - if (!versioningFailed) + if (!versioningFailed && poisonEvents.Count() == 0) { // In this case we skip the orchestration's execution since all tasks have been completed and it is in a terminal state. // Instead we "rewind" its execution by removing all failed tasks (see ProcessRewindOrchestrationDecision). @@ -873,14 +914,34 @@ await this.dispatchPipeline.RunAsync(dispatchContext, _ => /// The name of the dispatcher, used for tracing. /// The error propagation mode. /// The log helper. - /// True if workItem should be processed further. False otherwise. - internal static bool ReconcileMessagesWithState(TaskOrchestrationWorkItem workItem, string dispatcher, ErrorPropagationMode errorPropagationMode, LogHelper logHelper) + /// The poison message handler, or null if poison message handling is not enabled. + /// Indicates whether the work item is for an entity. + /// : The result of reconciling the work item messages with the orchestration state. + internal static async Task ReconcileMessagesWithStateAsync( + TaskOrchestrationWorkItem workItem, + string dispatcher, + ErrorPropagationMode errorPropagationMode, + LogHelper logHelper, + IPoisonMessageHandler? poisonMessageHandler, + bool isEntity) { foreach (TaskMessage message in workItem.NewMessages) { OrchestrationInstance orchestrationInstance = message.OrchestrationInstance; if (string.IsNullOrWhiteSpace(orchestrationInstance?.InstanceId)) { + string errorMessage = $"Work item includes a message with no orchestration instance ID with event type {message.Event.EventType}"; + if (poisonMessageHandler != null + && await poisonMessageHandler.HandleInvalidWorkItemAsync( + workItem, + PoisonMessageReason.MissingOrchestrationInstanceId, + errorMessage, + isEntity)) + { + logHelper.PoisonMessageDetected(workItem.OrchestrationRuntimeState.OrchestrationInstance, message.Event, errorMessage); + return WorkItemReconcileResult.PoisonMessageHandled; + } + throw TraceHelper.TraceException( TraceEventType.Error, $"{dispatcher}-OrchestrationInstanceMissing", @@ -890,7 +951,20 @@ internal static bool ReconcileMessagesWithState(TaskOrchestrationWorkItem workIt if (!workItem.OrchestrationRuntimeState.IsValid) { // we get here if the orchestration history is somehow corrupted (partially deleted, etc.) - return false; + string corruptionType = workItem.OrchestrationRuntimeState.Events.Count == 1 ? + $"its history contains exactly one event which is neither an {EventType.ExecutionStarted} or " + + $"{EventType.OrchestratorStarted} but rather has type {workItem.OrchestrationRuntimeState.Events[0].EventType}" : + $"its history contains multiple events but no {EventType.ExecutionStarted} event"; + if (poisonMessageHandler != null + && await poisonMessageHandler.HandleInvalidWorkItemAsync( + workItem, + PoisonMessageReason.InvalidRuntimeState, + $"Orchestration runtime state is invalid: {corruptionType}", + isEntity)) + { + return WorkItemReconcileResult.PoisonMessageHandled; + } + return WorkItemReconcileResult.Error; } if (workItem.OrchestrationRuntimeState.Events.Count == 1 && message.Event.EventType != EventType.ExecutionStarted) @@ -898,23 +972,35 @@ internal static bool ReconcileMessagesWithState(TaskOrchestrationWorkItem workIt // we get here because of: // i) responses for scheduled tasks after the orchestrations have been completed // ii) responses for explicitly deleted orchestrations - return false; + if (poisonMessageHandler != null + && await poisonMessageHandler.HandleInvalidWorkItemAsync( + workItem, + PoisonMessageReason.MissingExecutionStartedEvent, + $"Orchestration contains no {EventType.ExecutionStarted} event in its history and did not receive one as part of its new messages.", + isEntity)) + { + return WorkItemReconcileResult.PoisonMessageHandled; + } + return WorkItemReconcileResult.Error; } if (message.Event.EventType == EventType.ExecutionRewound && workItem.OrchestrationRuntimeState.OrchestrationStatus != OrchestrationStatus.Running && workItem.NewMessages.Count > 1) { - foreach (TaskMessage droppedMessage in workItem.NewMessages) + string errorMessage = "Multiple messages sent to an instance that is attempting to rewind from a terminal state. " + + "The only message that can be sent in this case is the rewind request."; + if (poisonMessageHandler != null + && await poisonMessageHandler.HandleInvalidWorkItemAsync( + workItem, + PoisonMessageReason.InvalidRewindRequest, + errorMessage, + isEntity)) { - if (droppedMessage.Event.EventType != EventType.ExecutionRewound) - { - logHelper.DroppingOrchestrationMessage(workItem, droppedMessage, "Multiple messages sent to an instance " + - "that is attempting to rewind from a terminal state. The only message that can be sent in " + - "this case is the rewind request."); - } + logHelper.PoisonMessageDetected(workItem.OrchestrationRuntimeState.OrchestrationInstance, message.Event, errorMessage); + return WorkItemReconcileResult.PoisonMessageHandled; } - return false; + return WorkItemReconcileResult.Error; } logHelper.ProcessingOrchestrationMessage(workItem, message); @@ -1005,7 +1091,14 @@ internal static bool ReconcileMessagesWithState(TaskOrchestrationWorkItem workIt } } - return true; + return WorkItemReconcileResult.Success; + } + + internal enum WorkItemReconcileResult + { + Success, + PoisonMessageHandled, + Error } TaskMessage? ProcessWorkflowCompletedTaskDecision( diff --git a/src/DurableTask.Core/Tracing/TraceHelper.cs b/src/DurableTask.Core/Tracing/TraceHelper.cs index 532e67a56..a912887fe 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(); } } }