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 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,20 @@ 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>
+        /// Gets or sets whether or not this event has been marked as "poisoned".
+        /// This can occur if the event's dispatch count exceeds a certain threshold,
+        /// or if some other error occurs during dispatch.
+        /// </summary>
+        [DataMember]
+        public bool IsPoisoned { get; set; }
+
         /// <summary>
         /// Implementation for <see cref="IExtensibleDataObject.ExtensionData"/>.
         /// </summary>

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,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.PoisonMessageDetected(
+                    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
     {
@@ -760,6 +762,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" 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,
+                    details));
+            }
+        }
         #endregion
 
         internal void OrchestrationDebugTrace(string instanceId, string executionId, string details)

src/DurableTask.Core/Logging/StructuredEventSource.cs

@@ -656,6 +656,31 @@ internal void DiscardingMessage(
             }
         }
 
+        [Event(EventIds.PoisonMessageDetected, Level = EventLevel.Error, Version = 1)]
+        internal void PoisonMessageDetected(
+            string InstanceId,
+            string ExecutionId,
+            string EventType,
+            int TaskEventId,
+            string Details,
+            string AppName,
+            string ExtensionVersion)
+        {
+            if (this.IsEnabled(EventLevel.Error))
+            {
+                // TODO: Use WriteEventCore for better performance
+                this.WriteEvent(
+                    EventIds.PoisonMessageDetected,
+                    InstanceId ?? string.Empty,
+                    ExecutionId ?? string.Empty,
+                    EventType,
+                    TaskEventId,
+                    Details,
+                    AppName,
+                    ExtensionVersion);
+            }
+        }
+
         [Event(EventIds.EntityBatchExecuting, Level = EventLevel.Informational, Version = 1)]
         internal void EntityBatchExecuting(
             string InstanceId,