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

Author: YunchuWang

.gitignore

@@ -351,3 +351,6 @@ MigrationBackup/
 
 # Rider (cross platform .NET/C# tools) working folder
 .idea/
+AzuriteConfig
+__azurite_db_*
+__blobstorage__

samples/LargePayloadConsoleApp/Program.cs

@@ -0,0 +1,29 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+namespace Microsoft.DurableTask;
+
+/// <summary>
+/// Exception thrown when a payload storage operation fails permanently.
+/// </summary>
+public sealed class PayloadStorageException : InvalidOperationException
+{
+    /// <summary>
+    /// Initializes a new instance of the <see cref="PayloadStorageException" /> class.
+    /// </summary>
+    /// <param name="message">The error message.</param>
+    public PayloadStorageException(string message)
+        : base(message)
+    {
+    }
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="PayloadStorageException" /> class.
+    /// </summary>
+    /// <param name="message">The error message.</param>
+    /// <param name="innerException">The inner exception.</param>
+    public PayloadStorageException(string message, Exception innerException)
+        : base(message, innerException)
+    {
+    }
+}

src/Extensions/AzureBlobPayloads/Exceptions/PayloadStorageException.cs

@@ -1,6 +1,7 @@
 // Copyright (c) Microsoft Corporation.
 // Licensed under the MIT License.
 
+using Azure;
 using Grpc.Core.Interceptors;
 
 using P = Microsoft.DurableTask.Protobuf;
@@ -39,10 +40,56 @@ protected override async Task ExternalizeRequestPayloadsAsync<TRequest>(TRequest
                 r.Input = await this.MaybeExternalizeAsync(r.Input, cancellation);
                 break;
             case P.ActivityResponse r:
-                r.Result = await this.MaybeExternalizeAsync(r.Result, cancellation);
+                try
+                {
+                    r.Result = await this.MaybeExternalizeAsync(r.Result, cancellation);
+                }
+                catch (Exception ex) when (IsPermanentStorageFailure(ex))
+                {
+                    // Permanent failure (e.g., payload exceeds configured maximum, 4xx auth/permission error).
+                    // Replace with a failure response so the orchestration sees a failed activity
+                    // instead of the work item being abandoned and redelivered indefinitely.
+                    r.Result = null;
+                    r.FailureDetails = new P.TaskFailureDetails
+                    {
+                        ErrorType = ex.GetType().FullName,
+                        ErrorMessage = ex.Message,
+                        StackTrace = ex.StackTrace,
+                        IsNonRetriable = true,
+                    };
+                }
+
                 break;
             case P.OrchestratorResponse r:
-                await this.ExternalizeOrchestratorResponseAsync(r, cancellation);
+                try
+                {
+                    await this.ExternalizeOrchestratorResponseAsync(r, cancellation);
+                }
+                catch (Exception ex) when (IsPermanentStorageFailure(ex))
+                {
+                    // Permanent failure during orchestration response externalization.
+                    // Replace all actions with a single Failed completion so the orchestration
+                    // terminates instead of being abandoned and redelivered indefinitely.
+                    r.Actions.Clear();
+                    r.CustomStatus = null;
+                    r.IsPartial = false;
+                    r.ChunkIndex = null;
+                    r.Actions.Add(new P.OrchestratorAction
+                    {
+                        CompleteOrchestration = new P.CompleteOrchestrationAction
+                        {
+                            OrchestrationStatus = P.OrchestrationStatus.Failed,
+                            FailureDetails = new P.TaskFailureDetails
+                            {
+                                ErrorType = ex.GetType().FullName,
+                                ErrorMessage = ex.Message,
+                                StackTrace = ex.StackTrace,
+                                IsNonRetriable = true,
+                            },
+                        },
+                    });
+                }
+
                 break;
             case P.EntityBatchResult r:
                 await this.ExternalizeEntityBatchResultAsync(r, cancellation);
@@ -360,4 +407,31 @@ async Task ResolveEventPayloadsAsync(P.HistoryEvent e, CancellationToken cancell
                 break;
         }
     }
+
+    /// <summary>
+    /// Determines whether an exception represents a permanent storage failure that will never
+    /// succeed on retry, such as payload exceeding the configured maximum or 4xx HTTP errors
+    /// (authentication, authorization, not found).
+    /// </summary>
+    static bool IsPermanentStorageFailure(Exception ex)
+    {
+        if (ex is PayloadStorageException)
+        {
+            return true;
+        }
+
+        // Azure SDK retries 408 (Request Timeout) and 429 (Too Many Requests) automatically
+        // (see ResponseClassifier.IsRetriableResponse in Azure.Core). All other 4xx status codes
+        // are NOT retried, meaning the request is fundamentally invalid
+        // (e.g., 401 bad credentials, 403 missing RBAC role, 404 account/container not found).
+        // These will never succeed on retry with the same configuration.
+        if (ex is RequestFailedException rfe
+            && rfe.Status >= 400 && rfe.Status < 500
+            && rfe.Status != 408 && rfe.Status != 429)
+        {
+            return true;
+        }
+
+        return false;
+    }
 }

src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs

@@ -174,17 +174,17 @@ public override AsyncServerStreamingCall<TResponse> AsyncServerStreamingCall<TRe
         }
 
         int size = Encoding.UTF8.GetByteCount(value);
-        if (size < this.options.ExternalizeThresholdBytes)
+        if (size < this.options.ThresholdBytes)
         {
             return value;
         }
 
         // Enforce a hard cap to prevent unbounded payload sizes
-        if (size > this.options.MaxExternalizedPayloadBytes)
+        if (size > this.options.MaxPayloadBytes)
         {
-            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.");
+            throw new PayloadStorageException(
+                $"Payload size {size / 1024} KB exceeds the configured maximum of {this.options.MaxPayloadBytes / 1024} KB. " +
+                "Reduce the payload size or increase the max payload size limit.");
         }
 
         return await this.payloadStore.UploadAsync(value!, cancellation);

src/Extensions/AzureBlobPayloads/Interceptors/PayloadInterceptor.cs

@@ -26,7 +26,7 @@ namespace Microsoft.DurableTask;
 /// </summary>
 public sealed class LargePayloadStorageOptions
 {
-    int externalizeThresholdBytes = 900_000;
+    int thresholdBytes = 900_000;
 
     /// <summary>
     /// Initializes a new instance of the <see cref="LargePayloadStorageOptions"/> class.
@@ -63,20 +63,20 @@ public LargePayloadStorageOptions(Uri accountUri, TokenCredential credential)
     /// Gets or sets the threshold in bytes at which payloads are externalized. Default is 900_000 bytes.
     /// Value must not exceed 1 MiB (1,048,576 bytes).
     /// </summary>
-    public int ExternalizeThresholdBytes
+    public int ThresholdBytes
     {
-        get => this.externalizeThresholdBytes;
+        get => this.thresholdBytes;
         set
         {
             const int OneMiB = 1 * 1024 * 1024;
             if (value > OneMiB)
             {
                 throw new ArgumentOutOfRangeException(
-                    nameof(this.ExternalizeThresholdBytes),
-                    $"ExternalizeThresholdBytes cannot exceed 1 MiB ({OneMiB} bytes).");
+                    nameof(this.ThresholdBytes),
+                    $"Payload storage threshold cannot exceed 1 MiB ({OneMiB} bytes).");
             }
 
-            this.externalizeThresholdBytes = value;
+            this.thresholdBytes = value;
         }
     }
 
@@ -85,7 +85,7 @@ public int ExternalizeThresholdBytes
     /// Defaults to 10MB. Requests exceeding this limit will fail fast
     /// with a clear error to prevent unbounded payload growth and excessive storage/network usage.
     /// </summary>
-    public int MaxExternalizedPayloadBytes { get; set; } = 10 * 1024 * 1024;
+    public int MaxPayloadBytes { get; set; } = 10 * 1024 * 1024;
 
     /// <summary>
     /// Gets or sets the Azure Storage connection string to the customer's storage account.
@@ -114,5 +114,5 @@ public int ExternalizeThresholdBytes
     /// Gets or sets a value indicating whether payloads should be gzip-compressed when stored.
     /// Defaults to true for reduced storage and bandwidth.
     /// </summary>
-    public bool CompressPayloads { get; set; } = true;
+    public bool CompressionEnabled { get; set; } = true;
 }

src/Extensions/AzureBlobPayloads/Options/LargePayloadStorageOptions.cs

@@ -55,7 +55,7 @@ public BlobPayloadStore(LargePayloadStorageOptions options)
                 Mode = RetryMode.Exponential,
                 MaxRetries = MaxRetryAttempts,
                 Delay = TimeSpan.FromMilliseconds(BaseDelayMs),
-                MaxDelay = TimeSpan.FromSeconds(MaxDelayMs),
+                MaxDelay = TimeSpan.FromMilliseconds(MaxDelayMs),
                 NetworkTimeout = TimeSpan.FromMinutes(NetworkTimeoutMinutes),
             },
         };
@@ -79,7 +79,7 @@ public override async Task<string> UploadAsync(string payLoad, CancellationToken
         // Ensure container exists (idempotent)
         await this.containerClient.CreateIfNotExistsAsync(PublicAccessType.None, default, default, cancellationToken);
 
-        if (this.options.CompressPayloads)
+        if (this.options.CompressionEnabled)
         {
             BlobOpenWriteOptions writeOptions = new()
             {
@@ -138,7 +138,7 @@ public override async Task<string> DownloadAsync(string token, CancellationToken
         }
         catch (RequestFailedException ex) when (ex.Status == (int)HttpStatusCode.NotFound)
         {
-            throw new InvalidOperationException(
+            throw new PayloadStorageException(
                 $"The blob '{name}' was not found in container '{container}'. " +
                 "The payload may have been deleted or the container was never created.",
                 ex);

src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs

@@ -957,7 +957,9 @@ async Task CompleteOrchestratorTaskWithChunkingAsync(
                 return null;
             }
 
-            P.TaskFailureDetails? validationFailure = ValidateActionsSize(response.Actions, maxChunkBytes);
+            P.TaskFailureDetails? validationFailure = this.worker.grpcOptions.Capabilities.Contains(P.WorkerCapability.LargePayloads)
+                ? null
+                : ValidateActionsSize(response.Actions, maxChunkBytes);
             if (validationFailure != null)
             {
                 // Complete the orchestration with a failed status and failure details
@@ -991,7 +993,7 @@ static bool TryAddAction(
                 int maxChunkBytes)
             {
                 int actionSize = action.CalculateSize();
-                if (currentSize + actionSize > maxChunkBytes)
+                if (currentSize + actionSize > maxChunkBytes && currentSize > 0)
                 {
                     return false;
                 }
@@ -1014,6 +1016,7 @@ static bool TryAddAction(
             int actionsCompletedSoFar = 0, chunkIndex = 0;
             List<P.OrchestratorAction> allActions = response.Actions.ToList();
             bool isPartial = true;
+            bool isChunkedMode = false;
 
             while (isPartial)
             {
@@ -1024,7 +1027,6 @@ static bool TryAddAction(
                     CompletionToken = response.CompletionToken,
                     RequiresHistory = response.RequiresHistory,
                     NumEventsProcessed = 0,
-                    ChunkIndex = chunkIndex,
                 };
 
                 int chunkPayloadSize = 0;
@@ -1040,6 +1042,20 @@ static bool TryAddAction(
                 isPartial = actionsCompletedSoFar < allActions.Count;
                 chunkedResponse.IsPartial = isPartial;
 
+                // Only activate chunked mode when we actually need multiple chunks.
+                // A single oversized action that fits in one chunk (via TryAddAction allowing
+                // the first item in an empty chunk) should be sent as non-chunked to avoid
+                // backend issues with ChunkIndex=0 + IsPartial=false.
+                if (isPartial)
+                {
+                    isChunkedMode = true;
+                }
+
+                if (isChunkedMode)
+                {
+                    chunkedResponse.ChunkIndex = chunkIndex;
+                }
+
                 if (chunkIndex == 0)
                 {
                     // The first chunk preserves the original response's NumEventsProcessed value (null)

src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs

@@ -0,0 +1,70 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+namespace Microsoft.DurableTask.Grpc.Tests;
+
+public class LargePayloadStorageOptionsTests
+{
+    [Fact]
+    public void ThresholdBytes_ExactlyOneMiB_DoesNotThrow()
+    {
+        // Arrange
+        LargePayloadStorageOptions options = new();
+
+        // Act & Assert
+        options.ThresholdBytes = 1 * 1024 * 1024;
+        Assert.Equal(1 * 1024 * 1024, options.ThresholdBytes);
+    }
+
+    [Fact]
+    public void ThresholdBytes_ExceedsOneMiB_ThrowsArgumentOutOfRange()
+    {
+        // Arrange
+        LargePayloadStorageOptions options = new();
+
+        // Act & Assert
+        ArgumentOutOfRangeException ex = Assert.Throws<ArgumentOutOfRangeException>(
+            () => options.ThresholdBytes = (1 * 1024 * 1024) + 1);
+        Assert.Contains("Payload storage threshold", ex.Message);
+    }
+
+    [Fact]
+    public void ThresholdBytes_DefaultValue_Is900000()
+    {
+        // Arrange & Act
+        LargePayloadStorageOptions options = new();
+
+        // Assert
+        Assert.Equal(900_000, options.ThresholdBytes);
+    }
+
+    [Fact]
+    public void MaxPayloadBytes_DefaultValue_Is10MB()
+    {
+        // Arrange & Act
+        LargePayloadStorageOptions options = new();
+
+        // Assert
+        Assert.Equal(10 * 1024 * 1024, options.MaxPayloadBytes);
+    }
+
+    [Fact]
+    public void CompressionEnabled_DefaultValue_IsTrue()
+    {
+        // Arrange & Act
+        LargePayloadStorageOptions options = new();
+
+        // Assert
+        Assert.True(options.CompressionEnabled);
+    }
+
+    [Fact]
+    public void ContainerName_DefaultValue_IsDurabletaskPayloads()
+    {
+        // Arrange & Act
+        LargePayloadStorageOptions options = new();
+
+        // Assert
+        Assert.Equal("durabletask-payloads", options.ContainerName);
+    }
+}