Add poison message handling to Azure Storage

src/DurableTask.AzureStorage/AnalyticsEventSource.cs

@@ -277,7 +277,7 @@ public void DuplicateMessageDetected(
                 ExtensionVersion);
         }
 
-        [Event(EventIds.PoisonMessageDetected, Level = EventLevel.Warning, Version = 3)]
+        [Event(EventIds.PoisonMessageDetected, Level = EventLevel.Warning, Version = 4)]
         public void PoisonMessageDetected(
             string Account,
             string TaskHub,
@@ -288,6 +288,7 @@ public void PoisonMessageDetected(
             string ExecutionId,
             string PartitionId,
             long DequeueCount,
+            string BlobName,
             string AppName,
             string ExtensionVersion)
         {
@@ -302,6 +303,7 @@ public void PoisonMessageDetected(
                 ExecutionId ?? string.Empty,
                 PartitionId,
                 DequeueCount,
+                BlobName,
                 AppName,
                 ExtensionVersion);
         }

src/DurableTask.AzureStorage/AzureStorageOrchestrationServiceSettings.cs

@@ -15,7 +15,6 @@ namespace DurableTask.AzureStorage
 {
     using System;
     using System.Runtime.Serialization;
-    using Azure.Data.Tables;
     using DurableTask.AzureStorage.Logging;
     using DurableTask.AzureStorage.Partitioning;
     using DurableTask.Core;
@@ -311,5 +310,40 @@ internal LogHelper Logger
         /// table must now be read before processing every orchestration work item to obtain the latest etag.
         /// </remarks>
         public bool UseInstanceTableEtag { get; set; } = false;
+
+        /// <summary>
+        /// Gets or sets a flag whether poison message storage is enabled.
+        /// </summary>
+        /// <remarks>
+        /// <para>
+        /// If enabled, orchestration, entity, and activity messages that have been dequeued for processing
+        /// more than <see cref="MaxDequeueCount"/> times will be moved to poison message storage in Azure Blob
+        /// Storage and deleted from their source queue.
+        /// This may leave orchestrations permanently as <see cref="OrchestrationStatus.Running"/> (or any other
+        /// non-terminal state) if the message(s) necessary for them make to progress are deemed "poisoned" and deleted.
+        /// </para>
+        /// </remarks>
+        public bool IsPoisonMessageStorageEnabled { get; set; } = true;
+
+        /// <summary>
+        /// Gets or sets the number of times a message is dequeued for processing before it is considered "poisoned",
+        /// moved to poison message storage, and deleted from the source queue. The default value is 5,000, chosen with
+        /// the intent that only very pathological cases will be handled automatically.
+        /// </summary>
+        /// <remarks>
+        /// This setting is applicable when <see cref="IsPoisonMessageStorageEnabled"/> is set to <c>true</c>.
+        /// </remarks>
+        public long MaxDequeueCount { get; set; } = 5000;
+
+        /// <summary>
+        /// Gets or sets the Azure Blob Storage container name suffix to use for poison message storage.
+        /// A container will be created with this suffix in the format "{taskhubname}-{suffix}".
+        /// The container name must adhere to the Azure Blob Storage container naming rules:
+        /// https://learn.microsoft.com/en-us/rest/api/storageservices/Naming-and-Referencing-Containers--Blobs--and-Metadata#directory-names
+        /// In particular, this means the total length of the name must not exceed 63 characters, the name can only contain lowercase letters,
+        /// numbers, and hyphens, must start and end with a letter or number, and cannot contain consecutive hyphens.
+        /// If not specified, the default value "poison-messages" will be used.
+        /// </summary>
+        public string PoisonMessageStorageContainerNameSuffix { get; set; } = "poison-messages";
     }
 }

src/DurableTask.AzureStorage/Logging/LogEvents.cs

@@ -666,7 +666,8 @@ public PoisonMessageDetected(
                 string instanceId,
                 string executionId,
                 string partitionId,
-                long dequeueCount)
+                long dequeueCount,
+                string blobName)
             {
                 this.Account = account;
                 this.TaskHub = taskHub;
@@ -677,6 +678,7 @@ public PoisonMessageDetected(
                 this.ExecutionId = executionId;
                 this.PartitionId = partitionId;
                 this.DequeueCount = dequeueCount;
+                this.BlobName = blobName;
             }
 
             [StructuredLogField]
@@ -706,18 +708,27 @@ public PoisonMessageDetected(
             [StructuredLogField]
             public long DequeueCount { get; }
 
+            [StructuredLogField]
+            public string BlobName { get; }
+
             public override EventId EventId => new EventId(
                 EventIds.PoisonMessageDetected,
                 nameof(EventIds.PoisonMessageDetected));
 
             public override LogLevel Level => LogLevel.Warning;
 
-            protected override string CreateLogMessage() => string.Format(
-                "{0}: Message {1} with ID {2} has been dequeued {3} times and is now considered poison",
-                this.InstanceId,
-                GetEventDescription(this.EventType, this.TaskEventId),
-                this.MessageId,
-                this.DequeueCount);
+            protected override string CreateLogMessage()
+            {
+                string message = string.Format(
+                    "{0}: Message {1} with ID {2} has been dequeued {3} times and is now considered poison, stored in blob {4}",
+                    this.InstanceId,
+                    GetEventDescription(this.EventType, this.TaskEventId),
+                    this.MessageId,
+                    this.DequeueCount,
+                    this.BlobName);
+
+                return message;
+            }
 
             void IEventSourceEvent.WriteEventSource() => AnalyticsEventSource.Log.PoisonMessageDetected(
                 this.Account,
@@ -729,6 +740,7 @@ void IEventSourceEvent.WriteEventSource() => AnalyticsEventSource.Log.PoisonMess
                 this.ExecutionId,
                 this.PartitionId,
                 this.DequeueCount,
+                this.BlobName,
                 Utils.AppName,
                 Utils.ExtensionVersion);
         }

src/DurableTask.AzureStorage/Logging/LogHelper.cs

@@ -236,7 +236,8 @@ internal void PoisonMessageDetected(
             string instanceId,
             string executionId,
             string partitionId,
-            long dequeueCount)
+            long dequeueCount,
+            string blobName)
         {
             var logEvent = new LogEvents.PoisonMessageDetected(
                 account,
@@ -247,7 +248,8 @@ internal void PoisonMessageDetected(
                 instanceId,
                 executionId,
                 partitionId,
-                dequeueCount);
+                dequeueCount,
+                blobName);
             this.WriteStructuredLog(logEvent);
         }
 

src/DurableTask.AzureStorage/Messaging/ControlQueue.cs

@@ -125,12 +125,31 @@ await batch.ParallelForEachAsync(async delegate (QueueMessage queueMessage)
                                     0 /* TaskEventId */,
                                     e.ToString());
 
+                                if (await this.CheckForAndHandlePoisonMessageAsync(
+                                    blobNamePrefix: "instance-messages/",
+                                    queueMessage,
+                                    linkedCts.Token))
+                                {
+                                    return;
+                                }
+
                                 // Abandon the message so we can try it again later.
                                 // Note: We will fetch the message again from the queue before retrying, so no need to read the receipt
                                 _ = await this.AbandonMessageAsync(queueMessage);
                                 return;
                             }
 
+                            if (await this.CheckForAndHandlePoisonMessageAsync(
+                                blobNamePrefix: "instance-messages/",
+                                queueMessage,
+                                linkedCts.Token,
+                                messageData.TaskMessage.OrchestrationInstance,
+                                messageData.TaskMessage.Event.EventType.ToString(),
+                                Utils.GetTaskEventId(messageData.TaskMessage.Event)))
+                            {
+                                return;
+                            }
+
                             // Check to see whether we've already dequeued this message.
                             if (!this.stats.PendingOrchestratorMessages.TryAdd(queueMessage.MessageId, 1))
                             {