more

Author: YunchuWang

src/Client/Core/OrchestrationMetadata.cs

@@ -80,6 +80,12 @@ public OrchestrationMetadata(string name, string instanceId)
     /// <value>The serialized custom status or <c>null</c>.</value>
     public string? SerializedCustomStatus { get; init; }
 
+    /// <summary>
+    /// Gets a value indicating whether large payload support is enabled.
+    /// </summary>
+    /// <value><c>true</c> if large payload support is enabled; <c>false</c> otherwise.</value>
+    public bool EnableLargePayloadSupport { get; init; }
+
     /// <summary>
     /// Gets the tags associated with the orchestration instance.
     /// </summary>
@@ -141,6 +147,7 @@ public OrchestrationMetadata(string name, string instanceId)
                 "that are fetched with the option to include input data.");
         }
 
+        Check.ThrowIfLargePayloadEnabled(this.EnableLargePayloadSupport, nameof(this.ReadInputAs));
         return this.DataConverter.Deserialize<T>(this.SerializedInput);
     }
 
@@ -167,6 +174,7 @@ public OrchestrationMetadata(string name, string instanceId)
                 "that are fetched with the option to include output data.");
         }
 
+        Check.ThrowIfLargePayloadEnabled(this.EnableLargePayloadSupport, nameof(this.ReadOutputAs));
         return this.DataConverter.Deserialize<T>(this.SerializedOutput);
     }
 
@@ -193,6 +201,7 @@ public OrchestrationMetadata(string name, string instanceId)
                 + " objects that are fetched with the option to include input and output data.");
         }
 
+        Check.ThrowIfLargePayloadEnabled(this.EnableLargePayloadSupport, nameof(this.ReadCustomStatusAs));
         return this.DataConverter.Deserialize<T>(this.SerializedCustomStatus);
     }
 
@@ -220,7 +229,12 @@ public OrchestrationMetadata(string name, string instanceId)
                 "that are fetched with the option to include input data.");
         }
 
-        return await this.DataConverter.DeserializeAsync<T>(this.SerializedInput, cancellationToken);
+        if (this.EnableLargePayloadSupport)
+        {
+            return await this.DataConverter.DeserializeAsync<T>(this.SerializedInput, cancellationToken);
+        }
+
+        return this.DataConverter.Deserialize<T>(this.SerializedInput);
     }
 
     /// <summary>
@@ -247,7 +261,12 @@ public OrchestrationMetadata(string name, string instanceId)
                 "that are fetched with the option to include output data.");
         }
 
-        return await this.DataConverter.DeserializeAsync<T>(this.SerializedOutput, cancellationToken);
+        if (this.EnableLargePayloadSupport)
+        {
+            return await this.DataConverter.DeserializeAsync<T>(this.SerializedOutput, cancellationToken);
+        }
+
+        return this.DataConverter.Deserialize<T>(this.SerializedOutput);
     }
 
     /// <summary>
@@ -274,6 +293,11 @@ public OrchestrationMetadata(string name, string instanceId)
                 " objects that are fetched with the option to include input and output data.");
         }
 
+        if (this.EnableLargePayloadSupport)
+        {
+            return await this.DataConverter.DeserializeAsync<T>(this.SerializedCustomStatus, cancellationToken);
+        }
+
         return await this.DataConverter.DeserializeAsync<T>(this.SerializedCustomStatus, cancellationToken);
     }
 

src/Client/Core/SerializedData.cs

@@ -20,6 +20,11 @@ public sealed class SerializedData(string data, DataConverter? converter = null)
     /// </summary>
     public string Value { get; } = Check.NotNull(data);
 
+    /// <summary>
+    /// Gets a value indicating whether large payload support is enabled.
+    /// </summary>
+    public bool EnableLargePayloadSupport { get; init; }
+
     /// <summary>
     /// Gets the data converter.
     /// </summary>
@@ -30,7 +35,11 @@ public sealed class SerializedData(string data, DataConverter? converter = null)
     /// </summary>
     /// <typeparam name="T">The type to deserialize into.</typeparam>
     /// <returns>The deserialized type.</returns>
-    public T ReadAs<T>() => this.Converter.Deserialize<T>(this.Value);
+    public T ReadAs<T>()
+    {
+        Check.ThrowIfLargePayloadEnabled(this.EnableLargePayloadSupport, nameof(this.ReadAs));
+        return this.Converter.Deserialize<T>(this.Value);
+    }
 
     /// <summary>
     /// Deserializes the data into <typeparamref name="T"/>.
@@ -44,21 +53,29 @@ public sealed class SerializedData(string data, DataConverter? converter = null)
     /// </summary>
     /// <param name="data">The data to serialize.</param>
     /// <param name="converter">The data converter.</param>
+    /// <param name="enableLargePayloadSupport">Whether to use async serialization for large payloads.</param>
     /// <returns>Serialized data.</returns>
-    internal static SerializedData Create(object data, DataConverter? converter = null)
+    internal static SerializedData Create(object data, DataConverter? converter = null, bool enableLargePayloadSupport = false)
     {
         converter ??= JsonDataConverter.Default;
-        return new SerializedData(converter.Serialize(data), converter);
+        return new SerializedData(converter.Serialize(data), converter)
+        {
+            EnableLargePayloadSupport = enableLargePayloadSupport,
+        };
     }
 
     /// <summary>
     /// Creates a new instance of <see cref="SerializedData"/> from the specified data.
     /// </summary>
     /// <param name="data">The data to serialize.</param>
     /// <param name="converter">The data converter.</param>
+    /// <param name="enableLargePayloadSupport">Whether to use async serialization for large payloads.</param>
     /// <returns>Serialized data.</returns>
-    internal static async Task<SerializedData> CreateAsync(object data, DataConverter converter)
+    internal static async Task<SerializedData> CreateAsync(object data, DataConverter converter, bool enableLargePayloadSupport = false)
     {
-        return new SerializedData(await converter.SerializeAsync(data), converter);
+        return new SerializedData(await converter.SerializeAsync(data), converter)
+        {
+            EnableLargePayloadSupport = enableLargePayloadSupport,
+        };
     }
 }

src/Client/Grpc/GrpcDurableEntityClient.cs

@@ -30,7 +30,11 @@ class GrpcDurableEntityClient : DurableEntityClient
     /// <param name="logger">The logger for logging client requests.</param>
     /// <param name="enableLargePayloadSupport">Whether to use async serialization for large payloads.</param>
     public GrpcDurableEntityClient(
-        string name, DataConverter dataConverter, TaskHubSidecarServiceClient sidecarClient, ILogger logger, bool enableLargePayloadSupport = false)
+        string name,
+        DataConverter dataConverter,
+        TaskHubSidecarServiceClient sidecarClient,
+        ILogger logger,
+        bool enableLargePayloadSupport = false)
         : base(name)
     {
         this.dataConverter = dataConverter;
@@ -305,7 +309,10 @@ EntityMetadata ToEntityMetadata(P.EntityMetadata metadata, bool includeState)
         EntityInstanceId entityId = new(coreEntityId.Name, coreEntityId.Key);
         bool hasState = metadata.SerializedState != null;
 
-        SerializedData? data = (includeState && hasState) ? new(metadata.SerializedState!, this.dataConverter) : null;
+        SerializedData? data = (includeState && hasState)
+            ? new SerializedData(metadata.SerializedState!, this.dataConverter)
+                { EnableLargePayloadSupport = this.enableLargePayloadSupport }
+            : null;
         return new EntityMetadata(entityId, data)
         {
             LastModifiedTime = metadata.LastModifiedTime.ToDateTimeOffset(),

src/Client/Grpc/GrpcDurableTaskClient.cs

@@ -521,6 +521,7 @@ OrchestrationMetadata CreateMetadata(P.OrchestrationState state, bool includeInp
             SerializedCustomStatus = state.CustomStatus,
             FailureDetails = state.FailureDetails.ToTaskFailureDetails(),
             DataConverter = includeInputsAndOutputs ? this.DataConverter : null,
+            EnableLargePayloadSupport = this.SupportsAsyncSerialization,
             Tags = new Dictionary<string, string>(state.Tags),
         };
 

src/Client/OrchestrationServiceClientShim/ShimDurableEntityClient.cs

@@ -216,7 +216,12 @@ EntityMetadata<T> ConvertSync<T>(EntityBackendQueries.EntityMetadata metadata)
 
     EntityMetadata Convert(EntityBackendQueries.EntityMetadata metadata)
     {
-        SerializedData? data = metadata.SerializedState is null ? null : new(metadata.SerializedState, this.Converter);
+        SerializedData? data = metadata.SerializedState is null
+            ? null
+            : new SerializedData(metadata.SerializedState, this.Converter)
+            {
+                EnableLargePayloadSupport = this.SupportsAsyncSerialization,
+            };
         return new(new EntityInstanceId(metadata.EntityId.Name, metadata.EntityId.Key), data)
         {
             LastModifiedTime = metadata.LastModifiedTime,

src/Client/OrchestrationServiceClientShim/ShimDurableTaskClient.cs

@@ -336,6 +336,7 @@ public override async Task<string> RestartAsync(
             SerializedOutput = state.Output,
             SerializedCustomStatus = state.Status,
             FailureDetails = state.FailureDetails?.ConvertFromCore(),
+            EnableLargePayloadSupport = this.SupportsAsyncSerialization,
         };
     }