Add poison message handling to the dispatchers

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,70 @@
+//  ----------------------------------------------------------------------------------
+//  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>
+    /// 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 message in the case that a message 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="orchestrationInstance">The orchestration 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 reason the event is "poisoned".</param>
+        /// <returns>True if the poison message was successfully handled, otherwise false.</returns>
+        public Task<bool> HandlePoisonMessageAsync(OrchestrationInstance? orchestrationInstance, HistoryEvent historyEvent, string reason);
+
+        /// <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">Why the work item is invalid.</param>
+        /// <returns>True if the poison message was successfully handled, otherwise false.</returns>
+        public Task<bool> HandleInvalidWorkItemAsync(TaskOrchestrationWorkItem workItem, string reason);
+
+        /// <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">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, string reason);
+    }
+}

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

src/DurableTask.Core/Logging/LogEvents.cs

@@ -1976,5 +1976,59 @@ void IEventSourceEvent.WriteEventSource() =>
                     Utils.PackageVersion);
         }
 
+        /// <summary>
+        /// Log event representing the discarding of a "poison" message.
+        /// </summary>
+        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);
+        }
+
     }
 }

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);
             }
         }
+
+        /// <summary>
+        /// Logs that a "poison message" has been detected and is being dropped.
+        /// </summary>
+        /// <param name="orchestrationInstance">The orchestration instance this event was sent to.</param>
+        /// <param name="historyEvent">The "poisoned" event.</param>
+        /// <param name="details">Extra details related to the processing of this poison message.</param>
+        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));
+            }
+        }
+
+        /// <summary>
+        /// Logs that a "poison" entity request message has been detected and is being dropped.
+        /// </summary>
+        /// <param name="orchestrationInstance">The orchestration instance this event was sent to.</param>
+        /// <param name="requestMessage">The "poisoned" request message.</param>
+        /// <param name="details">Extra details related to the processing of this poison message.</param>
+        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));
+            }
+        }
+
+        /// <summary>
+        /// Logs that a "poison" entity lock release message has been detected and is being dropped.
+        /// </summary>
+        /// <param name="orchestrationInstance">The orchestration instance this event was sent to.</param>
+        /// <param name="releaseMessage">The "poisoned" release message.</param>
+        /// <param name="dispatchCount">The dispatch count of the release message.</param>
+        /// <param name="details">Extra details related to the processing of this poison message.</param>
+        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)

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,