Add poison message handling to the dispatchers

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
         /// </summary>
         public QueueClientMessageEncoding QueueClientMessageEncoding { get; set; } = QueueClientMessageEncoding.UTF8;
 
+        /// <summary>
+        /// Gets or sets a flag whether poison storage is enabled.
+        /// </summary>
+        /// <remarks>
+        /// <para>If enabled, orchestration, entity, and activity messages that have been dispatched for processing
+        /// more than <see cref="MaxDispatchCount"/> times will be "failed" and moved to the poison storage.
+        /// </para>
+        /// </remarks>
+        public bool IsPoisonMessageStorageEnabled { get; set; } = false;
+
+        /// <summary>
+        /// 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.
+        /// </summary>
+        /// <remarks>
+        /// This setting is applicable when <see cref="IsPoisonMessageStorageEnabled"/> is set to <c>true</c>.
+        /// </remarks>
+        public int MaxDispatchCount { get; set; } = 5;
+
+        /// <summary>
+        /// 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.
+        /// </summary>
+        public string? PoisonMessageStorageContainerNamePrefix { get; set; } = "durable-task-poison";
+
         /// <summary>
         /// When true, an etag is used when attempting to make instance table updates upon completing an orchestration work item.
         /// </summary>

src/DurableTask.Core/Entities/ClientEntityHelpers.cs

@@ -91,6 +91,135 @@ public static EntityMessageEvent EmitUnlockForOrphanedLock(OrchestrationInstance
             return new EntityMessageEvent(EntityMessageEventNames.ReleaseMessageEventName, message, targetInstance);
         }
 
+        /// <summary>
+        /// 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.
+        /// </summary>
+        /// <remarks>
+        /// 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
+        /// <see langword="null"/> for those (and for any request that cannot be decoded well enough to identify the
+        /// caller).
+        /// </remarks>
+        /// <param name="entityRequestInput">The serialized input of the poison entity request (the <c>Input</c> of the
+        /// <see cref="History.EventRaisedEvent"/>).</param>
+        /// <param name="errorType">The error type to record in the failure response.</param>
+        /// <param name="errorMessage">The human-readable reason for the failure.</param>
+        /// <returns>An <see cref="EntityMessageEvent"/> targeting the calling instance, or <see langword="null"/> if no
+        /// caller can or should be notified.</returns>
+        public static EntityMessageEvent? TryCreateEntityOperationFailedResponse(string? entityRequestInput, string errorType, string errorMessage)
+        {
+            if (string.IsNullOrEmpty(entityRequestInput))
+            {
+                return null;
+            }
+
+            RequestMessage? requestMessage = TryDecodeEntityMessage<RequestMessage>(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);
+        }
+
+        /// <summary>
+        /// 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.
+        /// </summary>
+        /// <remarks>
+        /// 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
+        /// <see langword="null"/> since there is no way to know which lock to release.
+        /// </remarks>
+        /// <param name="releaseMessageInput">The serialized input of the original release message.</param>
+        /// <param name="entityInstance">The entity instance whose lock should be released.</param>
+        /// <param name="parentInstanceId">The parent instance ID of the lock owner, or <see langword="null"/> if it cannot be determined.</param>
+        /// <returns>An <see cref="EntityMessageEvent"/> targeting the entity, or <see langword="null"/> if the lock
+        /// owner cannot be identified.</returns>
+        public static EntityMessageEvent? TryRecreateEntityUnlock(string? releaseMessageInput, OrchestrationInstance entityInstance, out string? parentInstanceId)
+        {
+            parentInstanceId = null;
+
+            if (string.IsNullOrEmpty(releaseMessageInput))
+            {
+                return null;
+            }
+
+            ReleaseMessage? releaseMessage = TryDecodeEntityMessage<ReleaseMessage>(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<T>(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;
+            }
+        }
+
         /// <summary>
         /// 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.

src/DurableTask.Core/Entities/EventFormat/RequestMessage.cs

@@ -122,6 +122,11 @@ internal class RequestMessage : EntityMessage
         /// </summary>
         public string? ClientSpanId { get; set; }
 
+        /// <summary>
+        /// The number of times this request has been dispatched.
+        /// </summary>
+        public int DispatchCount { get; set; }
+
         /// <inheritdoc/>
         public override string GetShortDescription()
         {

src/DurableTask.Core/Entities/OrchestrationEntityContext.cs

@@ -341,6 +341,16 @@ public void CompleteAcquire(OperationResult result, Guid criticalSectionId)
             this.lockAcquisitionPending = false;
         }
 
+        /// <summary>
+        /// Called when the entity lock acquisition fails.
+        /// </summary>
+        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)

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)
+        {
+        }
+
         /// <summary>
         /// Gets the event type
         /// </summary>

src/DurableTask.Core/History/HistoryEvent.cs

@@ -89,6 +89,12 @@ protected HistoryEvent(int eventId)
         [DataMember]
         public virtual EventType EventType { get; private set; }
 
+        /// <summary>
+        /// Gets or sets the number of times this event has been dispatched.
+        /// </summary>
+        [DataMember(EmitDefaultValue = false)]
+        public int DispatchCount { get; set; }
+
         /// <summary>
         /// Implementation for <see cref="IExtensibleDataObject.ExtensionData"/>.
         /// </summary>

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;
+
+    /// <summary>
+    /// Categorizes why a message or work item was considered "poisoned" so that an
+    /// <see cref="IPoisonMessageHandler"/> can decide how to handle it without parsing human-readable strings.
+    /// </summary>
+    public enum PoisonMessageReason
+    {
+        /// <summary>
+        /// A message or event could not be deserialized
+        /// </summary>
+        DeserializationError,
+
+        /// <summary>
+        /// A message was dispatched for processing more than <see cref="IPoisonMessageHandler.MaxDispatchCount"/>
+        /// times and is therefore considered "poisoned".
+        /// </summary>
+        DispatchCount,
+
+        /// <summary>
+        /// A message or work item did not contain the orchestration instance information required to process it.
+        /// </summary>
+        MissingOrchestrationInstanceId,
+
+        /// <summary>
+        /// The orchestration runtime state reconstructed from history was invalid (for example, corrupted or
+        /// partially deleted history) and could not be processed.
+        /// </summary>
+        InvalidRuntimeState,
+
+        /// <summary>
+        /// The orchestration history did not contain the required <see cref="History.ExecutionStartedEvent"/> and
+        /// did not receive one as part of its new messages.
+        /// </summary>
+        MissingExecutionStartedEvent,
+
+        /// <summary>
+        /// 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.
+        /// </summary>
+        InvalidRewindRequest,
+
+        /// <summary>
+        /// A work item contained an event whose type is not supported by the dispatcher that received it.
+        /// </summary>
+        WrongEventType,
+
+        /// <summary>
+        /// An activity work item's <see cref="History.TaskScheduledEvent"/> did not specify an activity name, so the
+        /// activity could not be dispatched.
+        /// </summary>
+        MissingActivityName,
+    }
+
+    /// <summary>
+    /// Provides extensibility points for detecting and handling "poison" messages and invalid work items
+    /// in the task dispatchers.
+    /// </summary>
+    public interface IPoisonMessageHandler
+    {
+        /// <summary>
+        /// The maximum dispatch count after which a message should be considered "poisoned" if it is dispatched again.
+        /// </summary>
+        public int MaxDispatchCount { get; }
+
+        /// <summary>
+        /// Invoked to handle a poison entity message in the case that it cannot necessarily
+        /// be "failed" by the dispatchers, so the <see cref="IPoisonMessageHandler"/> must
+        /// decide what to do.
+        /// </summary>
+        /// <remarks>
+        /// If this method returns false, the dispatcher should fall back to the default behavior
+        /// followed when poison message handling is not enabled.
+        /// </remarks>
+        /// <param name="entityInstance">The entity instance the event was sent to, or null
+        /// if this information is not available.</param>
+        /// <param name="historyEvent">The "poisoned" history event.</param>
+        /// <param name="reason">The category describing why the event is "poisoned".</param>
+        /// <param name="details">A human-readable description of why the event is "poisoned".</param>
+        /// <returns>True if the poison message was successfully handled, otherwise false.</returns>
+        public Task<bool> HandlePoisonEntityMessageAsync(OrchestrationInstance? entityInstance, HistoryEvent historyEvent, PoisonMessageReason reason, string details);
+
+        /// <summary>
+        /// Invoked to handle a work item that is invalid and cannot be processed at all.
+        /// </summary>
+        /// <remarks>
+        /// If this method returns false, the dispatcher should fall back to the default behavior
+        /// followed in the case of an invalid work item.
+        /// </remarks>
+        /// <param name="workItem">The work item that could not be processed.</param>
+        /// <param name="reason">The category describing why the work item is invalid.</param>
+        /// <param name="details">A human-readable description of why the work item is invalid.</param>
+        /// <param name="isEntity">Indicates whether the work item is for an entity.</param>
+        /// <returns>True if the poison message was successfully handled, otherwise false.</returns>
+        public Task<bool> HandleInvalidWorkItemAsync(TaskOrchestrationWorkItem workItem, PoisonMessageReason reason, string details, bool isEntity);
+
+        /// <summary>
+        /// Invoked to handle a work item that is invalid and cannot be processed at all.
+        /// </summary>
+        /// <remarks>
+        /// If this method returns false, the dispatcher should fall back to the default behavior
+        /// followed in the case of an invalid work item.
+        /// </remarks>
+        /// <param name="workItem">The work item that could not be processed.</param>
+        /// <param name="reason">The category describing why the work item is invalid.</param>
+        /// <param name="details">A human-readable description of why the work item is invalid.</param>
+        /// <returns>True if the poison message was successfully handled, otherwise false.</returns>
+        public Task<bool> HandleInvalidWorkItemAsync(TaskActivityWorkItem workItem, PoisonMessageReason reason, string details);
+    }
+}

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;
     }
 }