From 65c4a578151d2a993322fd92766882da55618bb9 Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Mon, 27 Jul 2026 13:38:24 -0700 Subject: [PATCH 1/3] Avoid duplicate inline queue serialization Reuse the MessageData JSON already produced for size selection when the payload remains inline. Add exact wire-equivalence and threshold-path regression coverage for both serialization modes and a custom binder. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 9509e194-db0b-4024-b7b0-26481278e0db --- src/DurableTask.AzureStorage/MessageManager.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/DurableTask.AzureStorage/MessageManager.cs b/src/DurableTask.AzureStorage/MessageManager.cs index 1fc6d0785..e5809f618 100644 --- a/src/DurableTask.AzureStorage/MessageManager.cs +++ b/src/DurableTask.AzureStorage/MessageManager.cs @@ -119,7 +119,7 @@ public async Task SerializeMessageDataAsync(MessageData messageData, Can return Utils.SerializeToJson(serializer, wrapperMessageData); } - return Utils.SerializeToJson(serializer, messageData); + return rawContent; } /// From 7740fe4a464888542814f7b83770af31108340e9 Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Mon, 27 Jul 2026 13:38:43 -0700 Subject: [PATCH 2/3] Add queue serialization regression coverage Verify exact inline payload equivalence, single-pass custom binder behavior, and the inline/blob boundary for both data-contract serialization modes. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 9509e194-db0b-4024-b7b0-26481278e0db --- .../MessageManagerTests.cs | 208 +++++++++++++++++- 1 file changed, 200 insertions(+), 8 deletions(-) diff --git a/test/DurableTask.AzureStorage.Tests/MessageManagerTests.cs b/test/DurableTask.AzureStorage.Tests/MessageManagerTests.cs index 008a9eac7..de86ca3d3 100644 --- a/test/DurableTask.AzureStorage.Tests/MessageManagerTests.cs +++ b/test/DurableTask.AzureStorage.Tests/MessageManagerTests.cs @@ -14,15 +14,20 @@ namespace DurableTask.AzureStorage.Tests { using DurableTask.AzureStorage.Storage; + using DurableTask.Core; using DurableTask.Core.History; using Microsoft.VisualStudio.TestTools.UnitTesting; using Newtonsoft.Json; using System; using System.Collections.Generic; + using System.Text; + using System.Threading.Tasks; [TestClass] public class MessageManagerTests { + const int MaxStorageQueuePayloadSizeInBytes = 45 * 1024; + [DataTestMethod] [DataRow("System.Collections.Generic.Dictionary`2[[System.String, System.Private.CoreLib],[System.String, System.Private.CoreLib]]")] [DataRow("System.Collections.Generic.Dictionary`2[[System.String, mscorlib],[System.String, mscorlib]]")] @@ -69,6 +74,102 @@ public void DeserializesCustomTypes() Assert.AreEqual("tagValue", startedEvent.Tags["tag1"]); } + [DataTestMethod] + [DataRow(false)] + [DataRow(true)] + public async Task SerializeMessageDataAsync_InlinePayloadMatchesPreviousWirePayload(bool useDataContractSerialization) + { + var binder = new CustomNameTypeBinder(); + var settings = new AzureStorageOrchestrationServiceSettings + { + CustomMessageTypeBinder = binder, + UseDataContractSerialization = useDataContractSerialization, + }; + var manager = SetupMessageManager(settings, "$root"); + MessageData message = CreateMessageData("inline payload"); + + string actualPayload = await manager.SerializeMessageDataAsync(message); + string previousWirePayload = SerializeUsingMessageSettings( + new AzureStorageOrchestrationServiceSettings + { + CustomMessageTypeBinder = new CustomNameTypeBinder(), + UseDataContractSerialization = useDataContractSerialization, + }, + message); + + Assert.AreEqual(previousWirePayload, actualPayload); + Assert.AreEqual(Encoding.UTF8.GetByteCount(previousWirePayload), message.TotalMessageSizeBytes); + Assert.AreEqual(MessageFormatFlags.InlineJson, manager.GetMessageFormatFlags(message)); + Assert.IsNull(message.CompressedBlobName); + Assert.AreEqual(1, binder.MessageDataSerializationCount); + StringAssert.Contains(actualPayload, CustomNameTypeBinder.CustomAssemblyName); + + MessageData roundTrippedMessage = manager.DeserializeMessageData(actualPayload); + Assert.AreEqual(message.ActivityId, roundTrippedMessage.ActivityId); + Assert.AreEqual( + ((GenericEvent)message.TaskMessage.Event).Data, + ((GenericEvent)roundTrippedMessage.TaskMessage.Event).Data); + } + + [DataTestMethod] + [DataRow(false, 0)] + [DataRow(false, 1)] + [DataRow(true, 0)] + [DataRow(true, 1)] + public async Task SerializeMessageDataAsync_UsesExpectedStoragePathAtThreshold( + bool useDataContractSerialization, + int bytesOverThreshold) + { + var settings = new AzureStorageOrchestrationServiceSettings + { + UseDataContractSerialization = useDataContractSerialization, + }; + string containerName = $"message-manager-{Guid.NewGuid():N}"; + var manager = SetupMessageManager(settings, containerName); + int targetSize = MaxStorageQueuePayloadSizeInBytes + bytesOverThreshold; + MessageData message = CreateMessageDataWithSerializedSize(settings, targetSize); + string fullMessagePayload = SerializeUsingMessageSettings(settings, message); + + try + { + string queuePayload = await manager.SerializeMessageDataAsync(message); + MessageFormatFlags expectedFormat = bytesOverThreshold == 0 + ? MessageFormatFlags.InlineJson + : MessageFormatFlags.StorageBlob; + + Assert.AreEqual(targetSize, message.TotalMessageSizeBytes); + Assert.AreEqual(expectedFormat, manager.GetMessageFormatFlags(message)); + + if (expectedFormat == MessageFormatFlags.InlineJson) + { + Assert.AreEqual(fullMessagePayload, queuePayload); + Assert.IsNull(message.CompressedBlobName); + } + else + { + string expectedWrapperPayload = SerializeUsingMessageSettings( + settings, + new MessageData { CompressedBlobName = message.CompressedBlobName }); + Assert.AreEqual(expectedWrapperPayload, queuePayload); + + MessageData wrapper = manager.DeserializeMessageData(queuePayload); + Assert.IsNull(wrapper.TaskMessage); + Assert.IsFalse(string.IsNullOrWhiteSpace(wrapper.CompressedBlobName)); + Assert.AreEqual(message.CompressedBlobName, wrapper.CompressedBlobName); + + string storedPayload = await manager.DownloadAndDecompressAsBytesAsync(wrapper.CompressedBlobName); + Assert.AreEqual(fullMessagePayload, storedPayload); + } + } + finally + { + if (bytesOverThreshold > 0) + { + await manager.DeleteContainerAsync(); + } + } + } + [DataTestMethod] [DataRow("blob.bin", "blob.bin")] [DataRow("@#$%!", "%40%23%24%25%21")] @@ -94,17 +195,108 @@ private string GetMessage(string dictionaryType) private MessageManager SetupMessageManager(ICustomTypeBinder binder) { - var azureStorageClient = new AzureStorageClient( - new AzureStorageOrchestrationServiceSettings - { - StorageAccountClientProvider = new StorageAccountClientProvider("UseDevelopmentStorage=true"), - }); - - return new MessageManager( + return SetupMessageManager( new AzureStorageOrchestrationServiceSettings { CustomMessageTypeBinder = binder }, - azureStorageClient, "$root"); } + + private static MessageManager SetupMessageManager( + AzureStorageOrchestrationServiceSettings settings, + string containerName) + { + settings.StorageAccountClientProvider = + new StorageAccountClientProvider(TestHelpers.GetTestStorageAccountConnectionString()); + var azureStorageClient = new AzureStorageClient(settings); + return new MessageManager(settings, azureStorageClient, containerName); + } + + private static MessageData CreateMessageData(string payload) + { + var orchestrationInstance = new OrchestrationInstance + { + InstanceId = "message-manager-instance", + ExecutionId = "message-manager-execution", + }; + var taskMessage = new TaskMessage + { + Event = new GenericEvent(1, payload), + OrchestrationInstance = orchestrationInstance, + SequenceNumber = 42, + }; + + return new MessageData( + taskMessage, + Guid.Parse("55f31df8-9abb-4d86-a197-78fd0908efcf"), + "message-manager-queue", + orchestrationEpisode: 3, + sender: orchestrationInstance) + { + SequenceNumber = 43, + SerializableTraceContext = "trace-context", + }; + } + + private static MessageData CreateMessageDataWithSerializedSize( + AzureStorageOrchestrationServiceSettings settings, + int targetSize) + { + MessageData message = CreateMessageData(string.Empty); + int emptyPayloadSize = Encoding.UTF8.GetByteCount(SerializeUsingMessageSettings(settings, message)); + Assert.IsTrue(targetSize >= emptyPayloadSize); + + ((GenericEvent)message.TaskMessage.Event).Data = new string('x', targetSize - emptyPayloadSize); + int actualSize = Encoding.UTF8.GetByteCount(SerializeUsingMessageSettings(settings, message)); + Assert.AreEqual(targetSize, actualSize); + return message; + } + + private static string SerializeUsingMessageSettings( + AzureStorageOrchestrationServiceSettings settings, + MessageData message) + { + var serializerSettings = new JsonSerializerSettings + { + TypeNameHandling = TypeNameHandling.Objects, + SerializationBinder = new TypeNameSerializationBinder(settings.CustomMessageTypeBinder), + }; + + if (settings.UseDataContractSerialization) + { + serializerSettings.Converters.Add(new DataContractJsonConverter()); + } + + return Utils.SerializeToJson(JsonSerializer.Create(serializerSettings), message); + } + } + + internal class CustomNameTypeBinder : ICustomTypeBinder + { + public const string CustomAssemblyName = "MessageManagerTests.CustomAssembly"; + readonly Dictionary serializedTypes = new Dictionary(); + + public int MessageDataSerializationCount { get; private set; } + + public void BindToName(Type serializedType, out string assemblyName, out string typeName) + { + assemblyName = CustomAssemblyName; + typeName = serializedType.FullName!; + this.serializedTypes[typeName] = serializedType; + + if (serializedType == typeof(MessageData)) + { + this.MessageDataSerializationCount++; + } + } + + public Type BindToType(string assemblyName, string typeName) + { + if (this.serializedTypes.TryGetValue(typeName, out Type? serializedType)) + { + return serializedType; + } + + throw new JsonSerializationException($"Unknown serialized type '{typeName}'."); + } } internal class KnownTypeBinder : ICustomTypeBinder From f09bb565d85eb517bcd63a3ce53c682115f02b89 Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Mon, 21 Sep 2026 17:17:17 -0700 Subject: [PATCH 3/3] Document queue serialization compatibility expectations Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- docs/providers/azure-storage.md | 9 +++++++++ .../AzureStorageOrchestrationServiceSettings.cs | 6 +++++- src/DurableTask.AzureStorage/ICustomTypeBinder.cs | 13 +++++++++---- 3 files changed, 23 insertions(+), 5 deletions(-) diff --git a/docs/providers/azure-storage.md b/docs/providers/azure-storage.md index cc9960838..a17444727 100644 --- a/docs/providers/azure-storage.md +++ b/docs/providers/azure-storage.md @@ -152,6 +152,15 @@ The Azure Storage provider creates these resources: | **Partitions Table** | `{taskhub}Partitions` | Partition leases (table manager) | | **Lease Blobs** | `{taskhub}-leases/` | Partition leases (blob manager) | +### Queue Message Serialization + +The provider serializes the full `MessageData` envelope to JSON and measures its UTF-8 byte count before any Base64 encoding. Messages at or below 45 KiB (46,080 bytes) are sent inline using that same JSON. Larger messages are compressed and uploaded to blob storage; the queue receives a separately serialized blob-reference wrapper. + +This transport serialization is separate from the application input/output serialization performed by `DataConverter`. `AzureStorageOrchestrationServiceSettings.CustomMessageTypeBinder` configures type binding for queue messages. Its `ICustomTypeBinder.BindToName` implementation should return stable names for a given type. Serialization invocation counts are implementation details, not an API contract; custom binders, getters, and serialization callbacks on transport types must not depend on a particular number of passes for correctness. + +> [!NOTE] +> The inline path reuses the JSON produced for the size check instead of serializing the full message a second time. This preserves the wire payload for stable messages with deterministic, side-effect-free serialization. Custom serialization logic that changes its output or relies on side effects across successive passes can observe a behavior change and should be reviewed when upgrading. + ### Partitioning The Azure Storage provider uses **partitions** to distribute orchestration workloads across workers. Each partition corresponds to exactly one **control queue**. diff --git a/src/DurableTask.AzureStorage/AzureStorageOrchestrationServiceSettings.cs b/src/DurableTask.AzureStorage/AzureStorageOrchestrationServiceSettings.cs index bc5689e0e..5def0af60 100644 --- a/src/DurableTask.AzureStorage/AzureStorageOrchestrationServiceSettings.cs +++ b/src/DurableTask.AzureStorage/AzureStorageOrchestrationServiceSettings.cs @@ -239,8 +239,12 @@ public class AzureStorageOrchestrationServiceSettings public bool DisableExecutionStartedDeduplication { get; set; } /// - /// Gets or sets an optional custom type binder used when trying to deserialize queued messages. + /// Gets or sets an optional custom type binder used when serializing and deserializing queued messages. /// + /// + /// Custom bindings must be stable and must not depend on serialization invocation counts. + /// See for details. + /// public ICustomTypeBinder? CustomMessageTypeBinder { get; set; } /// diff --git a/src/DurableTask.AzureStorage/ICustomTypeBinder.cs b/src/DurableTask.AzureStorage/ICustomTypeBinder.cs index 44e85fdd3..4f9b90b16 100644 --- a/src/DurableTask.AzureStorage/ICustomTypeBinder.cs +++ b/src/DurableTask.AzureStorage/ICustomTypeBinder.cs @@ -17,15 +17,20 @@ namespace DurableTask.AzureStorage { /// /// Abstraction to prevent surfacing the dynamic loading between System.Runtime.Serialization.SerializationBinder and Newtonsoft.Json.Serialization.ISerializationBinder - /// Used when deserializing QueueMessages to MessageData to allow providing custom type binding. - /// Does not support custom bindings for DurableTask types + /// Used when serializing and deserializing queue messages to allow providing custom type binding. + /// Custom bindings for DurableTask types are not used during deserialization. /// public interface ICustomTypeBinder { /// - /// When implemented, controls the binding of a serialized object to a type. + /// When implemented, supplies the assembly and type names written during serialization. /// - /// The type of the object the formatter creates a new instance of. + /// + /// Return stable names for a given type. The number of invocations is an implementation detail + /// and can vary with the message size, storage path, and serializer configuration. + /// Implementations must not depend on a particular invocation count for correctness. + /// + /// The type of the object being serialized. /// Specifies the System.Reflection.Assembly name of the serialized object. /// Specifies the System.Type name of the serialized object. void BindToName(Type serializedType, out string assemblyName, out string typeName);