fix: improve large payload error handling — better error message and prevent infinite retry and fix conflict with auto chunking

src/Extensions/AzureBlobPayloads/Interceptors/PayloadInterceptor.cs

@@ -68,6 +68,14 @@ Status GetStatus()
 
             if (startCallTask.IsFaulted)
             {
+                // Distinguish permanent failures from transient ones so callers
+                // can avoid retrying errors that will never succeed.
+                Exception? inner = startCallTask.Exception?.InnerException ?? startCallTask.Exception;
+                if (IsPermanentFailure(inner))
+                {
+                    return new Status(StatusCode.FailedPrecondition, inner?.Message ?? "Permanent failure");
+                }
+
                 return new Status(StatusCode.Internal, startCallTask.Exception?.Message ?? "Unknown error");
             }
 
@@ -80,6 +88,28 @@ Status GetStatus()
             return new Status(StatusCode.Unknown, string.Empty);
         }
 
+        static bool IsPermanentFailure(Exception? ex)
+        {
+            // InvalidOperationException: payload too large, blob not found, decompression error, etc.
+            if (ex is InvalidOperationException or ArgumentException)
+            {
+                return true;
+            }
+
+            // Azure Storage errors that are permanent (auth/permission failures).
+            // Transient errors (429, 500, 503) are already retried by the Azure SDK
+            // client with exponential backoff before reaching here, so if they still
+            // fail, they may also be effectively permanent — but we keep them as
+            // Internal/retryable to give the upper layers a chance.
+            if (ex is Azure.RequestFailedException rfe &&
+                rfe.Status is 401 or 403 or 404)
+            {
+                return true;
+            }
+
+            return false;
+        }
+
         Metadata GetTrailers()
         {
             return startCallTask.Status == TaskStatus.RanToCompletion ? startCallTask.Result.GetTrailers() : [];
@@ -183,8 +213,11 @@ public override AsyncServerStreamingCall<TResponse> AsyncServerStreamingCall<TRe
         if (size > this.options.MaxExternalizedPayloadBytes)
         {
             throw new InvalidOperationException(
-                $"Payload size {size / 1024} kb exceeds the configured maximum of {this.options.MaxExternalizedPayloadBytes / 1024} kb. " +
-                "Consider reducing the payload or increase MaxExternalizedPayloadBytes setting.");
+                $"Payload size {size / 1024} kb exceeds the configured maximum of " +
+                $"{this.options.MaxExternalizedPayloadBytes / 1024} kb. Consider reducing the payload size, " +
+                $"or increase the max payload size limit " +
+                $"(LargePayloadStorageOptions.MaxExternalizedPayloadBytes for standalone SDK, " +
+                $"or 'largePayloadStorageMaxPayloadBytes' in host.json extensions.durableTask section for Azure Functions).");
         }
 
         return await this.payloadStore.UploadAsync(value!, cancellation);

src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs

@@ -762,7 +762,8 @@ await this.client.AbandonTaskOrchestratorWorkItemAsync(
             await this.CompleteOrchestratorTaskWithChunkingAsync(
                 response,
                 this.worker.grpcOptions.CompleteOrchestrationWorkItemChunkSizeInBytes,
-                cancellationToken);
+                cancellationToken,
+                completionToken);
         }
 
         async Task OnRunActivityAsync(P.ActivityRequest request, string completionToken, CancellationToken cancellation)
@@ -853,7 +854,33 @@ await this.client.AbandonTaskActivityWorkItemAsync(
             // Stop the trace activity here to avoid including the completion time in the latency calculation
             traceActivity?.Stop();
 
-            await this.client.CompleteActivityTaskAsync(response, cancellationToken: cancellation);
+            try
+            {
+                await this.client.CompleteActivityTaskAsync(response, cancellationToken: cancellation);
+            }
+            catch (RpcException ex) when (ex.StatusCode == StatusCode.FailedPrecondition)
+            {
+                // Permanent failure (e.g., payload too large for externalization).
+                // Complete the activity with a failure instead of letting the exception propagate
+                // to RunBackgroundTask, which would abandon and cause an infinite re-delivery loop.
+                this.Logger.UnexpectedError(ex, instance.InstanceId);
+
+                P.ActivityResponse failureResponse = new()
+                {
+                    InstanceId = instance.InstanceId,
+                    TaskId = request.TaskId,
+                    FailureDetails = new P.TaskFailureDetails
+                    {
+                        ErrorType = typeof(InvalidOperationException).FullName,
+                        ErrorMessage = ex.Status.Detail,
+                        StackTrace = ex.ToString(),
+                        IsNonRetriable = true,
+                    },
+                    CompletionToken = completionToken,
+                };
+
+                await this.client.CompleteActivityTaskAsync(failureResponse, cancellationToken: cancellation);
+            }
         }
 
         async Task OnRunEntityBatchAsync(
@@ -919,7 +946,31 @@ async Task OnRunEntityBatchAsync(
                 completionToken,
                 operationInfos?.Take(batchResult.Results?.Count ?? 0));
 
-            await this.client.CompleteEntityTaskAsync(response, cancellationToken: cancellation);
+            try
+            {
+                await this.client.CompleteEntityTaskAsync(response, cancellationToken: cancellation);
+            }
+            catch (RpcException ex) when (ex.StatusCode == StatusCode.FailedPrecondition)
+            {
+                // Permanent failure (e.g., payload too large for externalization).
+                // Complete with a batch-level failure using failureDetails so the entity
+                // framework sees the failure instead of an infinite abandon/re-deliver loop.
+                this.Logger.UnexpectedError(ex, batchRequest.InstanceId ?? string.Empty);
+
+                P.EntityBatchResult failureResponse = new()
+                {
+                    CompletionToken = completionToken ?? response.CompletionToken,
+                    FailureDetails = new P.TaskFailureDetails
+                    {
+                        ErrorType = typeof(InvalidOperationException).FullName,
+                        ErrorMessage = ex.Status.Detail,
+                        StackTrace = ex.ToString(),
+                        IsNonRetriable = true,
+                    },
+                };
+
+                await this.client.CompleteEntityTaskAsync(failureResponse, cancellationToken: cancellation);
+            }
         }
 
         /// <summary>
@@ -931,7 +982,8 @@ async Task OnRunEntityBatchAsync(
         async Task CompleteOrchestratorTaskWithChunkingAsync(
             P.OrchestratorResponse response,
             int maxChunkBytes,
-            CancellationToken cancellationToken)
+            CancellationToken cancellationToken,
+            string? completionToken = null)
         {
             // Validate that no single action exceeds the maximum chunk size
             static P.TaskFailureDetails? ValidateActionsSize(IEnumerable<P.OrchestratorAction> actions, int maxChunkBytes)
@@ -1002,6 +1054,8 @@ static bool TryAddAction(
             }
 
             // Check if the entire response fits in one chunk
+            try
+            {
             int totalSize = response.CalculateSize();
             if (totalSize <= maxChunkBytes)
             {
@@ -1054,6 +1108,38 @@ static bool TryAddAction(
                 // Send the chunk
                 await this.client.CompleteOrchestratorTaskAsync(chunkedResponse, cancellationToken: cancellationToken);
             }
+            }
+            catch (RpcException ex) when (ex.StatusCode == StatusCode.FailedPrecondition)
+            {
+                // Permanent failure (e.g., payload too large for externalization).
+                // Complete with a failure to prevent infinite re-delivery loop.
+                this.Logger.UnexpectedError(ex, response.InstanceId);
+
+                P.OrchestratorResponse failureResponse = new()
+                {
+                    InstanceId = response.InstanceId,
+                    CompletionToken = response.CompletionToken ?? completionToken,
+                    Actions =
+                    {
+                        new P.OrchestratorAction
+                        {
+                            CompleteOrchestration = new P.CompleteOrchestrationAction
+                            {
+                                OrchestrationStatus = P.OrchestrationStatus.Failed,
+                                FailureDetails = new P.TaskFailureDetails
+                                {
+                                    ErrorType = typeof(InvalidOperationException).FullName,
+                                    ErrorMessage = ex.Status.Detail,
+                                    StackTrace = ex.ToString(),
+                                    IsNonRetriable = true,
+                                },
+                            },
+                        },
+                    },
+                };
+
+                await this.client.CompleteOrchestratorTaskAsync(failureResponse, cancellationToken: cancellationToken);
+            }
         }
     }
 }