Isolate inbound payload hydration failures to their deliveries

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## Unreleased
 
+- Isolate incoming payload hydration failures to the failed delivery, preserving the shared work-item stream and unrelated in-flight work ([#825](https://github.com/microsoft/durabletask-dotnet/issues/825)).
 - Externalize continue-as-new carryover event payloads using the configured UTF-8 size limits, preventing oversized completion responses ([#819](https://github.com/microsoft/durabletask-dotnet/issues/819)).
 
 ## v1.27.1

src/Extensions/AzureBlobPayloads/Interceptors/PayloadInterceptor.cs

@@ -4,6 +4,7 @@
 using System.Text;
 using Grpc.Core;
 using Grpc.Core.Interceptors;
+using P = Microsoft.DurableTask.Protobuf;
 
 namespace Microsoft.DurableTask;
 
@@ -143,7 +144,20 @@ public override AsyncServerStreamingCall<TResponse> AsyncServerStreamingCall<TRe
 
         IAsyncStreamReader<TResponse> wrapped = new TransformingStreamReader<TResponse>(call.ResponseStream, async (msg, ct) =>
         {
-            await this.ResolveResponsePayloadsAsync(msg, ct);
+            try
+            {
+                await this.ResolveResponsePayloadsAsync(msg, ct);
+            }
+            catch (Exception ex) when (msg is P.WorkItem workItem
+                && workItem.RequestCase is P.WorkItem.RequestOneofCase.ActivityRequest
+                    or P.WorkItem.RequestOneofCase.OrchestratorRequest
+                    or P.WorkItem.RequestOneofCase.EntityRequest
+                    or P.WorkItem.RequestOneofCase.EntityRequestV2
+                && !ct.IsCancellationRequested)
+            {
+                throw new WorkItemPayloadException(workItem, ex);
+            }
+
             return msg;
         });
 

src/Grpc/Grpc.csproj

@@ -21,4 +21,10 @@
     <Protobuf Include="**/*.proto" GrpcServices="Both" Access="Public"  />
   </ItemGroup>
 
+  <ItemGroup>
+    <InternalsVisibleTo Include="Microsoft.DurableTask.Worker.Grpc" Key="$(StrongNamePublicKey)" />
+    <InternalsVisibleTo Include="Microsoft.DurableTask.Extensions.AzureBlobPayloads" Key="$(StrongNamePublicKey)" />
+    <InternalsVisibleTo Include="Microsoft.DurableTask.Worker.Grpc.Tests" Key="$(StrongNamePublicKey)" />
+  </ItemGroup>
+
 </Project>

src/Grpc/WorkItemPayloadException.cs

@@ -0,0 +1,20 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+using P = Microsoft.DurableTask.Protobuf;
+
+namespace Microsoft.DurableTask;
+
+/// <summary>
+/// A received delivery whose payloads could not be resolved. The item is only safe to abandon, not dispatch.
+/// </summary>
+/// <param name="workItem">The failed delivery, including its original completion token.</param>
+/// <param name="innerException">The payload resolution failure.</param>
+internal sealed class WorkItemPayloadException(P.WorkItem workItem, Exception innerException)
+    : Exception("Failed to resolve payloads for a work-item delivery.", innerException)
+{
+    /// <summary>
+    /// Gets the failed delivery. Its payload fields may be partially resolved.
+    /// </summary>
+    public P.WorkItem WorkItem { get; } = workItem;
+}

src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs

@@ -279,6 +279,13 @@ static OrchestrationRuntimeState BuildRuntimeState(
             return runtimeState;
         }
 
+        static string GetWorkItemInstanceId(P.WorkItem? workItem) =>
+            workItem?.OrchestratorRequest?.InstanceId ??
+            workItem?.ActivityRequest?.OrchestrationInstance?.InstanceId ??
+            workItem?.EntityRequest?.InstanceId ??
+            workItem?.EntityRequestV2?.InstanceId ??
+            string.Empty;
+
         async ValueTask<IReadOnlyList<P.HistoryEvent>> GetPastEventsAsync(
             P.OrchestratorRequest orchestratorRequest,
             CancellationToken cancellation)
@@ -385,16 +392,15 @@ async Task ProcessWorkItemsAsync(
             // window. If that fires we treat the stream as silently disconnected and reconnect.
             TimeSpan silentDisconnectTimeout = this.internalOptions.SilentDisconnectTimeout;
 
-            // NOTE: the consumer deliberately does NOT wrap its await foreach in an outer loop.
-            // The underlying IAsyncStreamReader is single-use — once the server terminates the stream
-            // (e.g. via a graceful HTTP/2 GOAWAY with OK trailers during a rolling upgrade), MoveNext
-            // returns false forever and re-entering await foreach would tight-spin with no yield.
+            // Read directly so a delivery-local hydration failure does not terminate a ReadAllAsync
+            // iterator. A false MoveNext still ends consumption permanently (no drain tight-spin).
             WorkItemStreamResult result = await WorkItemStreamConsumer.ConsumeAsync(
-                ct => stream.ResponseStream.ReadAllAsync(ct),
+                stream.ResponseStream,
                 silentDisconnectTimeout,
                 workItem => this.DispatchWorkItem(workItem, cancellation),
                 onFirstMessage,
-                cancellation);
+                cancellation,
+                onPayloadFailure: this.OnWorkItemPayloadFailureAsync);
 
             switch (result.Outcome)
             {
@@ -534,90 +540,82 @@ void RunBackgroundTask(P.WorkItem? workItem, Func<Task> handler, CancellationTok
                 }
                 catch (Exception ex)
                 {
-                    string instanceId =
-                        workItem?.OrchestratorRequest?.InstanceId ??
-                        workItem?.ActivityRequest?.OrchestrationInstance?.InstanceId ??
-                        workItem?.EntityRequest?.InstanceId ??
-                        workItem?.EntityRequestV2?.InstanceId ??
-                        string.Empty;
-                    this.Logger.UnexpectedError(ex, instanceId);
-
-                    if (workItem?.OrchestratorRequest != null)
-                    {
-                        this.Logger.AbandoningOrchestratorWorkItem(instanceId, workItem.CompletionToken ?? string.Empty);
-                        await this.ExecuteWithRetryAsync(
-                            async () => await this.client.AbandonTaskOrchestratorWorkItemAsync(
-                                new P.AbandonOrchestrationTaskRequest
-                                {
-                                    CompletionToken = workItem.CompletionToken,
-                                },
-                                cancellationToken: cancellation),
-                            nameof(this.client.AbandonTaskOrchestratorWorkItemAsync),
-                            cancellation);
-                        this.Logger.AbandonedOrchestratorWorkItem(instanceId, workItem.CompletionToken ?? string.Empty);
-                    }
-                    else if (workItem?.ActivityRequest != null)
-                    {
-                        this.Logger.AbandoningActivityWorkItem(
-                            instanceId,
-                            workItem.ActivityRequest.Name,
-                            workItem.ActivityRequest.TaskId,
-                            workItem.CompletionToken ?? string.Empty);
-                        await this.ExecuteWithRetryAsync(
-                            async () => await this.client.AbandonTaskActivityWorkItemAsync(
-                                new P.AbandonActivityTaskRequest
-                                {
-                                    CompletionToken = workItem.CompletionToken,
-                                },
-                                cancellationToken: cancellation),
-                            nameof(this.client.AbandonTaskActivityWorkItemAsync),
-                            cancellation);
-                        this.Logger.AbandonedActivityWorkItem(
-                           instanceId,
-                           workItem.ActivityRequest.Name,
-                           workItem.ActivityRequest.TaskId,
-                           workItem.CompletionToken ?? string.Empty);
-                    }
-                    else if (workItem?.EntityRequest != null)
-                    {
-                        this.Logger.AbandoningEntityWorkItem(
-                            workItem.EntityRequest.InstanceId,
-                            workItem.CompletionToken ?? string.Empty);
-                        await this.ExecuteWithRetryAsync(
-                            async () => await this.client.AbandonTaskEntityWorkItemAsync(
-                                new P.AbandonEntityTaskRequest
-                                {
-                                    CompletionToken = workItem.CompletionToken,
-                                },
-                                cancellationToken: cancellation),
-                            nameof(this.client.AbandonTaskEntityWorkItemAsync),
-                            cancellation);
-                        this.Logger.AbandonedEntityWorkItem(
-                            workItem.EntityRequest.InstanceId,
-                            workItem.CompletionToken ?? string.Empty);
-                    }
-                    else if (workItem?.EntityRequestV2 != null)
-                    {
-                        this.Logger.AbandoningEntityWorkItem(
-                            workItem.EntityRequestV2.InstanceId,
-                            workItem.CompletionToken ?? string.Empty);
-                        await this.ExecuteWithRetryAsync(
-                            async () => await this.client.AbandonTaskEntityWorkItemAsync(
-                                new P.AbandonEntityTaskRequest
-                                {
-                                    CompletionToken = workItem.CompletionToken,
-                                },
-                                cancellationToken: cancellation),
-                            nameof(this.client.AbandonTaskEntityWorkItemAsync),
-                            cancellation);
-                        this.Logger.AbandonedEntityWorkItem(
-                            workItem.EntityRequestV2.InstanceId,
-                            workItem.CompletionToken ?? string.Empty);
-                    }
+                    this.Logger.UnexpectedError(ex, GetWorkItemInstanceId(workItem));
+                    await this.AbandonWorkItemAsync(workItem, cancellation);
                 }
             });
         }
 
+        async Task OnWorkItemPayloadFailureAsync(WorkItemPayloadException exception, CancellationToken cancellation)
+        {
+            P.WorkItem workItem = exception.WorkItem;
+            this.Logger.UnexpectedError(exception, GetWorkItemInstanceId(workItem));
+            try
+            {
+                await this.AbandonWorkItemAsync(workItem, cancellation);
+            }
+            catch (RpcException ex) when (ex.StatusCode == StatusCode.NotFound)
+            {
+                // Ownership loss is delivery-local; other abandon failures must reach the reconnect loop.
+                this.Logger.WorkItemOwnershipLost(workItem.RequestCase.ToString(), workItem.CompletionToken, ex);
+            }
+        }
+
+        async Task AbandonWorkItemAsync(P.WorkItem? workItem, CancellationToken cancellation)
+        {
+            string instanceId = GetWorkItemInstanceId(workItem);
+            if (workItem?.OrchestratorRequest != null)
+            {
+                this.Logger.AbandoningOrchestratorWorkItem(instanceId, workItem.CompletionToken ?? string.Empty);
+                await this.ExecuteWithRetryAsync(
+                    async () => await this.client.AbandonTaskOrchestratorWorkItemAsync(
+                        new P.AbandonOrchestrationTaskRequest
+                        {
+                            CompletionToken = workItem.CompletionToken,
+                        },
+                        cancellationToken: cancellation),
+                    nameof(this.client.AbandonTaskOrchestratorWorkItemAsync),
+                    cancellation);
+                this.Logger.AbandonedOrchestratorWorkItem(instanceId, workItem.CompletionToken ?? string.Empty);
+            }
+            else if (workItem?.ActivityRequest != null)
+            {
+                this.Logger.AbandoningActivityWorkItem(
+                    instanceId,
+                    workItem.ActivityRequest.Name,
+                    workItem.ActivityRequest.TaskId,
+                    workItem.CompletionToken ?? string.Empty);
+                await this.ExecuteWithRetryAsync(
+                    async () => await this.client.AbandonTaskActivityWorkItemAsync(
+                        new P.AbandonActivityTaskRequest
+                        {
+                            CompletionToken = workItem.CompletionToken,
+                        },
+                        cancellationToken: cancellation),
+                    nameof(this.client.AbandonTaskActivityWorkItemAsync),
+                    cancellation);
+                this.Logger.AbandonedActivityWorkItem(
+                   instanceId,
+                   workItem.ActivityRequest.Name,
+                   workItem.ActivityRequest.TaskId,
+                   workItem.CompletionToken ?? string.Empty);
+            }
+            else if (workItem?.EntityRequest != null || workItem?.EntityRequestV2 != null)
+            {
+                this.Logger.AbandoningEntityWorkItem(instanceId, workItem.CompletionToken ?? string.Empty);
+                await this.ExecuteWithRetryAsync(
+                    async () => await this.client.AbandonTaskEntityWorkItemAsync(
+                        new P.AbandonEntityTaskRequest
+                        {
+                            CompletionToken = workItem.CompletionToken,
+                        },
+                        cancellationToken: cancellation),
+                    nameof(this.client.AbandonTaskEntityWorkItemAsync),
+                    cancellation);
+                this.Logger.AbandonedEntityWorkItem(instanceId, workItem.CompletionToken ?? string.Empty);
+            }
+        }
+
         async Task OnRunOrchestratorAsync(
             P.OrchestratorRequest request,
             string completionToken,

src/Worker/Grpc/Logs.cs

@@ -107,5 +107,8 @@ static partial class Logs
 
         [LoggerMessage(EventId = 78, Level = LogLevel.Warning, Message = "Activity notification callback failed for phase '{Phase}'.")]
         public static partial void ActivityNotificationFailed(this ILogger logger, ActivityNotificationPhase phase, Exception exception);
+
+        [LoggerMessage(EventId = 79, Level = LogLevel.Warning, Message = "Ownership of '{workItemType}' work item was lost before abandonment. Completion token = '{completionToken}'")]
+        public static partial void WorkItemOwnershipLost(this ILogger logger, string workItemType, string completionToken, Exception exception);
     }
 }

src/Worker/Grpc/WorkItemStreamConsumer.cs

@@ -43,10 +43,7 @@ internal static class WorkItemStreamConsumer
     /// <summary>
     /// Consume a work-item stream until shutdown, silent disconnect, or graceful drain.
     /// </summary>
-    /// <param name="openStream">
-    /// Factory that opens the stream with the supplied linked-cancellation token. Production passes
-    /// <c>ct => stream.ResponseStream.ReadAllAsync(ct)</c>; tests pass arbitrary fakes.
-    /// </param>
+    /// <param name="stream">The response reader. Each read receives the linked cancellation token.</param>
     /// <param name="silentDisconnectTimeout">
     /// How long to wait between successive items (or health pings) before treating the stream as
     /// silently disconnected. Non-positive values disable detection entirely.
@@ -68,14 +65,19 @@ internal static class WorkItemStreamConsumer
     /// real elapsed time. Never invoked when <paramref name="silentDisconnectTimeout"/> disables
     /// detection.
     /// </param>
+    /// <param name="onPayloadFailure">
+    /// Optional delivery-local recovery for a payload resolution failure. Exceptions from recovery
+    /// propagate to the caller for connection-level handling.
+    /// </param>
     /// <returns>The classified outcome plus whether any message was observed.</returns>
     public static async Task<WorkItemStreamResult> ConsumeAsync(
-        Func<CancellationToken, IAsyncEnumerable<P.WorkItem>> openStream,
+        IAsyncStreamReader<P.WorkItem> stream,
         TimeSpan silentDisconnectTimeout,
         Action<P.WorkItem> onItem,
         Action? onFirstMessage,
         CancellationToken cancellation,
-        Action? onSilentDisconnectTimerArmed = null)
+        Action? onSilentDisconnectTimerArmed = null,
+        Func<WorkItemPayloadException, CancellationToken, Task>? onPayloadFailure = null)
     {
         bool silentDisconnectEnabled = silentDisconnectTimeout > TimeSpan.Zero;
 
@@ -100,19 +102,38 @@ void ArmSilentDisconnectTimer()
 
         bool firstMessageObserved = false;
 
-        try
+        void ObserveMessage()
         {
-            await foreach (P.WorkItem workItem in openStream(tokenSource.Token).ConfigureAwait(false))
+            ArmSilentDisconnectTimer();
+
+            if (!firstMessageObserved)
             {
-                ArmSilentDisconnectTimer();
+                firstMessageObserved = true;
+                onFirstMessage?.Invoke();
+            }
+        }
 
-                if (!firstMessageObserved)
+        try
+        {
+            while (true)
+            {
+                try
                 {
-                    firstMessageObserved = true;
-                    onFirstMessage?.Invoke();
+                    if (!await stream.MoveNext(tokenSource.Token).ConfigureAwait(false))
+                    {
+                        break;
+                    }
                 }
+                catch (WorkItemPayloadException ex) when (onPayloadFailure is not null)
+                {
+                    ObserveMessage();
+                    await onPayloadFailure(ex, tokenSource.Token).ConfigureAwait(false);
+                    continue;
+                }
+
+                ObserveMessage();
 
-                onItem(workItem);
+                onItem(stream.Current);
             }
         }
         catch (OperationCanceledException) when (cancellation.IsCancellationRequested)

src/Worker/Grpc/Worker.Grpc.csproj

@@ -22,6 +22,7 @@
   </ItemGroup>
 
   <ItemGroup>
+    <InternalsVisibleTo Include="Microsoft.DurableTask.Grpc.IntegrationTests" Key="$(StrongNamePublicKey)" />
     <!-- The AzureBlobPayloads auto-purge activities share this worker's gRPC transport, so their tests drive a
          real worker through a channel recreate. Reaching that path quickly needs the internal connect-loop
          knobs (recreate threshold), which are deliberately not part of any supported surface. -->