Add Poison Message Handling to the Dispatchers

src/DurableTask.AzureServiceFabric/FabricOrchestrationService.cs

@@ -473,6 +473,8 @@ public Task ReleaseTaskOrchestrationWorkItemAsync(TaskOrchestrationWorkItem work
 
         public BehaviorOnContinueAsNew EventBehaviourForContinueAsNew { get; } = BehaviorOnContinueAsNew.Ignore;
 
+        public int MaxDispatchCount => int.MaxValue;
+
         // Note: Do not rely on cancellationToken parameter to this method because the top layer does not yet implement any cancellation.
         public async Task<TaskActivityWorkItem> LockNextTaskActivityWorkItem(TimeSpan receiveTimeout, CancellationToken cancellationToken)
         {

src/DurableTask.AzureStorage/AzureStorageOrchestrationService.cs

@@ -2169,6 +2169,14 @@ public bool UseSeparateQueuesForEntityWorkItems
             set => this.settings.UseSeparateQueueForEntityWorkItems = value;
         }
 
+        /// <summary>
+        /// The maximum amount of times a message can be dispatched before it is considered "poisoned" and
+        /// moved to poison storage. 
+        /// Currently this storage provider does not have poison message handling so this value is set to the
+        /// maximum integer value.
+        /// </summary>
+        public int MaxDispatchCount => int.MaxValue;
+
         /// <summary>
         /// Disposes of the current object.
         /// </summary>

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

@@ -122,6 +122,11 @@ internal class RequestMessage : EntityMessage
         /// </summary>
         public string? ClientSpanId { get; set; }
 
+        /// <summary>
+        /// The dispatch count of this request message.
+        /// </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/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]
+        public int DispatchCount { get; set; }
+
         /// <summary>
         /// Implementation for <see cref="IExtensibleDataObject.ExtensionData"/>.
         /// </summary>

src/DurableTask.Core/IOrchestrationService.cs

@@ -102,6 +102,12 @@ public interface IOrchestrationService
         /// </summary>
         BehaviorOnContinueAsNew EventBehaviourForContinueAsNew { get; }
 
+        /// <summary>
+        /// Gets the maximum amount of times the same event can be dispatched before it is considered "poisonous"
+        /// and the corresponding operation is failed.
+        /// </summary>
+        int MaxDispatchCount { get; }
+
         /// <summary>
         ///     Wait for the next orchestration work item and return the orchestration work item
         /// </summary>

src/DurableTask.Core/Logging/EventIds.cs

@@ -71,5 +71,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

@@ -1945,5 +1945,54 @@ 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, string details)
+            {
+                this.InstanceId = orchestrationInstance?.InstanceId ?? string.Empty;
+                this.ExecutionId = orchestrationInstance?.ExecutionId ?? string.Empty;
+                this.EventType = eventType;
+                this.TaskEventId = taskEventId;
+                this.Details = details;
+            }
+
+            [StructuredLogField]
+            public string InstanceId { get; }
+
+            [StructuredLogField]
+            public string ExecutionId { get; }
+
+            [StructuredLogField]
+            public string EventType { get; }
+
+            [StructuredLogField]
+            public int TaskEventId { get; }
+
+            [StructuredLogField]
+            public string Details { get; }
+
+            public override EventId EventId => new EventId(
+                EventIds.PoisonMessageDetected,
+                nameof(EventIds.PoisonMessageDetected));
+
+            public override LogLevel Level => LogLevel.Error;
+
+            protected override string CreateLogMessage() =>
+                $"{this.InstanceId}: Poison message detected for {GetEventDescription(this.EventType, this.TaskEventId)}: {this.Details}";
+
+            void IEventSourceEvent.WriteEventSource() =>
+                StructuredEventSource.Log.DiscardingMessage(
+                    this.InstanceId,
+                    this.ExecutionId,
+                    this.EventType,
+                    this.TaskEventId,
+                    this.Details,
+                    Utils.AppName,
+                    Utils.PackageVersion);
+        }
+
     }
 }

src/DurableTask.Core/Logging/LogHelper.cs

@@ -13,13 +13,15 @@
 #nullable enable
 namespace DurableTask.Core.Logging
 {
-    using System;
-    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;
+    using System;
+    using System.Collections.Generic;
+    using System.Text;
 
     class LogHelper
     {
@@ -745,6 +747,42 @@ 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),
+                    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" reuest 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,
+                    details));
+            }
+        }
         #endregion
 
         internal void OrchestrationDebugTrace(string instanceId, string executionId, string details)