Add structured abandonment details to Azure Storage logs

src/DurableTask.AzureStorage/AnalyticsEventSource.cs

@@ -165,7 +165,7 @@ public void DeletingMessage(
                 ExtensionVersion);
         }
 
-        [Event(EventIds.AbandoningMessage, Level = EventLevel.Warning, Version = 7)]
+        [Event(EventIds.AbandoningMessage, Level = EventLevel.Warning, Version = 8)]
         public void AbandoningMessage(
             string Account,
             string TaskHub,
@@ -179,7 +179,8 @@ public void AbandoningMessage(
             string PopReceipt,
             int VisibilityTimeoutSeconds,
             string AppName,
-            string ExtensionVersion)
+            string ExtensionVersion,
+            string Details)
         {
             this.WriteEvent(
                 EventIds.AbandoningMessage,
@@ -195,7 +196,8 @@ public void AbandoningMessage(
                 PopReceipt ?? string.Empty,
                 VisibilityTimeoutSeconds,
                 AppName,
-                ExtensionVersion);
+                ExtensionVersion,
+                Details ?? string.Empty);
         }
 
         [Event(EventIds.AssertFailure, Level = EventLevel.Warning, Message = "An unexpected condition was detected: {2}", Version = 2)]

src/DurableTask.AzureStorage/AzureStorageOrchestrationService.cs

@@ -726,7 +726,9 @@ async Task<TaskOrchestrationWorkItem> LockNextTaskOrchestrationWorkItemAsync(boo
                     // Make sure we still own the partition. If not, abandon the session.
                     if (session.ControlQueue.IsReleased)
                     {
-                        await this.AbandonAndReleaseSessionAsync(session);
+                        await this.AbandonAndReleaseSessionAsync(
+                            session,
+                            "The control queue was released.");
                         return null;
                     }
 
@@ -771,13 +773,18 @@ async Task<TaskOrchestrationWorkItem> LockNextTaskOrchestrationWorkItemAsync(boo
                     if (outOfOrderMessages?.Count > 0)
                     {
                         // This will also remove the messages from the current batch.
-                        await this.AbandonMessagesAsync(session, outOfOrderMessages);
+                        await this.AbandonMessagesAsync(
+                            session,
+                            outOfOrderMessages,
+                            "Message was received out of order.");
                     }
 
                     if (session.CurrentMessageBatch.Count == 0)
                     {
                         // All messages were removed. Release the work item.
-                        await this.AbandonAndReleaseSessionAsync(session);
+                        await this.AbandonAndReleaseSessionAsync(
+                            session,
+                            "No processable messages remained in the session.");
                         return null;
                     }
 
@@ -870,7 +877,9 @@ async Task<TaskOrchestrationWorkItem> LockNextTaskOrchestrationWorkItemAsync(boo
                     if (session != null)
                     {
                         // host is shutting down - release any queued messages
-                        await this.AbandonAndReleaseSessionAsync(session);
+                        await this.AbandonAndReleaseSessionAsync(
+                            session,
+                            "Message processing was canceled during shutdown or listener cancellation.");
                     }
 
                     return null;
@@ -1125,11 +1134,11 @@ await this.trackingStore.UpdateInstanceStatusForCompletedOrchestrationAsync(
             return null;
         }
 
-        async Task AbandonAndReleaseSessionAsync(OrchestrationSession session)
+        async Task AbandonAndReleaseSessionAsync(OrchestrationSession session, string details)
         {
             try
             {
-                await this.AbandonSessionAsync(session);
+                await this.AbandonSessionAsync(session, details);
             }
             finally
             {
@@ -1488,20 +1497,25 @@ public Task AbandonTaskOrchestrationWorkItemAsync(TaskOrchestrationWorkItem work
                 return Utils.CompletedTask;
             }
 
-            return this.AbandonSessionAsync(session);
+            return this.AbandonSessionAsync(
+                session,
+                "The orchestration work item was abandoned by the dispatcher.");
         }
 
-        Task AbandonSessionAsync(OrchestrationSession session)
+        Task AbandonSessionAsync(OrchestrationSession session, string details)
         {
             session.StartNewLogicalTraceScope();
-            return this.AbandonMessagesAsync(session, session.CurrentMessageBatch.ToList());
+            return this.AbandonMessagesAsync(session, session.CurrentMessageBatch.ToList(), details);
         }
 
-        async Task AbandonMessagesAsync(OrchestrationSession session, IList<MessageData> messages)
+        async Task AbandonMessagesAsync(
+            OrchestrationSession session,
+            IList<MessageData> messages,
+            string details)
         {
             await messages.ParallelForEachAsync(
                 this.settings.MaxStorageOperationConcurrency,
-                message => session.ControlQueue.AbandonMessageAsync(message, session));
+                message => session.ControlQueue.AbandonMessageAsync(message, details, session));
 
             // Remove the messages from the current batch. The remaining messages
             // may still be able to be processed
@@ -1680,7 +1694,10 @@ public async Task AbandonTaskActivityWorkItemAsync(TaskActivityWorkItem workItem
 
             session.StartNewLogicalTraceScope();
 
-            await this.workItemQueue.AbandonMessageAsync(session.MessageData, session);
+            await this.workItemQueue.AbandonMessageAsync(
+                session.MessageData,
+                "The activity work item was abandoned by the dispatcher.",
+                session);
 
             if (this.activeActivitySessions.TryRemove(workItem.Id, out _))
             {

src/DurableTask.AzureStorage/Logging/LogEvents.cs

@@ -336,7 +336,8 @@ public AbandoningMessage(
                 string partitionId,
                 long sequenceNumber,
                 string popReceipt,
-                int visibilityTimeoutSeconds)
+                int visibilityTimeoutSeconds,
+                string details)
             {
                 this.Account = account;
                 this.TaskHub = taskHub;
@@ -349,6 +350,7 @@ public AbandoningMessage(
                 this.SequenceNumber = sequenceNumber;
                 this.PopReceipt = popReceipt;
                 this.VisibilityTimeoutSeconds = visibilityTimeoutSeconds;
+                this.Details = details;
             }
 
             [StructuredLogField]
@@ -384,6 +386,9 @@ public AbandoningMessage(
             [StructuredLogField]
             public int VisibilityTimeoutSeconds { get; }
 
+            [StructuredLogField]
+            public string Details { get; }
+
             public override EventId EventId => new EventId(
                 EventIds.AbandoningMessage,
                 nameof(EventIds.AbandoningMessage));
@@ -410,7 +415,8 @@ void IEventSourceEvent.WriteEventSource() => AnalyticsEventSource.Log.Abandoning
                 this.PopReceipt,
                 this.VisibilityTimeoutSeconds,
                 Utils.AppName,
-                Utils.ExtensionVersion);
+                Utils.ExtensionVersion,
+                this.Details);
         }
 
         internal class AssertFailure : StructuredLogEvent, IEventSourceEvent

src/DurableTask.AzureStorage/Logging/LogHelper.cs

@@ -134,7 +134,8 @@ internal void AbandoningMessage(
             string partitionId,
             long sequenceNumber,
             string popReceipt,
-            int visibilityTimeoutSeconds)
+            int visibilityTimeoutSeconds,
+            string details)
         {
             var logEvent = new LogEvents.AbandoningMessage(
                 account,
@@ -147,7 +148,8 @@ internal void AbandoningMessage(
                 partitionId,
                 sequenceNumber,
                 popReceipt,
-                visibilityTimeoutSeconds);
+                visibilityTimeoutSeconds,
+                details);
             this.WriteStructuredLog(logEvent);
         }
 

src/DurableTask.AzureStorage/Messaging/ControlQueue.cs

@@ -127,7 +127,9 @@ await batch.ParallelForEachAsync(async delegate (QueueMessage queueMessage)
 
                                 // 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);
+                                _ = await this.AbandonMessageAsync(
+                                    queueMessage,
+                                    "Message deserialization failed.");
                                 return;
                             }
 
@@ -192,21 +194,25 @@ await batch.ParallelForEachAsync(async delegate (QueueMessage queueMessage)
         }
 
         // This overload is intended for cases where we aren't able to deserialize an instance of MessageData.
-        public Task<UpdateReceipt?> AbandonMessageAsync(QueueMessage queueMessage)
+        public Task<UpdateReceipt?> AbandonMessageAsync(QueueMessage queueMessage, string details)
         {
             this.stats.PendingOrchestratorMessages.TryRemove(queueMessage.MessageId, out _);
             return base.AbandonMessageAsync(
                 queueMessage,
                 taskMessage: null,
                 instance: null,
                 traceActivityId: null,
-                sequenceNumber: -1);
+                sequenceNumber: -1,
+                details: details);
         }
 
-        public override Task AbandonMessageAsync(MessageData message, SessionBase? session = null)
+        public override Task AbandonMessageAsync(
+            MessageData message,
+            string abandonmentDetails,
+            SessionBase? session = null)
         {
             this.stats.PendingOrchestratorMessages.TryRemove(message.OriginalQueueMessage.MessageId, out _);
-            return base.AbandonMessageAsync(message, session);
+            return base.AbandonMessageAsync(message, abandonmentDetails, session);
         }
 
         public override Task DeleteMessageAsync(MessageData message, SessionBase? session = null)

src/DurableTask.AzureStorage/Messaging/TaskHubQueue.cs

@@ -212,7 +212,10 @@ await this.storageQueue.AddMessageAsync(
             return initialVisibilityDelay;
         }
 
-        public virtual async Task AbandonMessageAsync(MessageData message, SessionBase? session = null)
+        public virtual async Task AbandonMessageAsync(
+            MessageData message,
+            string abandonmentDetails,
+            SessionBase? session = null)
         {
             QueueMessage queueMessage = message.OriginalQueueMessage;
             TaskMessage taskMessage = message.TaskMessage;
@@ -224,7 +227,8 @@ public virtual async Task AbandonMessageAsync(MessageData message, SessionBase?
                 taskMessage,
                 instance,
                 session?.TraceActivityId,
-                sequenceNumber);
+                sequenceNumber,
+                abandonmentDetails);
 
             // If we've successfully abandoned the message, update the pop receipt
             // (even though we'll likely no longer interact with this message)
@@ -239,7 +243,8 @@ public virtual async Task AbandonMessageAsync(MessageData message, SessionBase?
             TaskMessage? taskMessage,
             OrchestrationInstance? instance,
             Guid? traceActivityId,
-            long sequenceNumber)
+            long sequenceNumber,
+            string details)
         {
             string instanceId = instance?.InstanceId ?? string.Empty;
             string executionId = instance?.ExecutionId ?? string.Empty;
@@ -277,7 +282,8 @@ public virtual async Task AbandonMessageAsync(MessageData message, SessionBase?
                 this.storageQueue.Name,
                 sequenceNumber,
                 queueMessage.PopReceipt,
-                numSecondsToWait);
+                numSecondsToWait,
+                details);
 
             try
             {

src/DurableTask.AzureStorage/OrchestrationSessionManager.cs

@@ -326,7 +326,11 @@ async Task<IEnumerable<MessageData>> DedupeExecutionStartedMessagesAsync(
                 filteredMessages = filteredMessages.Except(messagesToDefer);
 
                 // Defer messages on a background thread to avoid blocking the dequeue loop
-                _ = Task.Run(() => messagesToDefer.ParallelForEachAsync(msg => controlQueue.AbandonMessageAsync(msg, session: null)));
+                _ = Task.Run(() => messagesToDefer.ParallelForEachAsync(
+                    msg => controlQueue.AbandonMessageAsync(
+                        msg,
+                        "Execution-start message was deferred pending instance status reconciliation.",
+                        session: null)));
             }
 
             if (messagesToDiscard?.Count > 0)