Support Configurable Queue Client Encoding Strategy and Cross-Encoding Handling

Test/DurableTask.AzureStorage.Tests/QueueClientEncodingStrategyIntegrationTests.cs

@@ -0,0 +1,269 @@
+//  ----------------------------------------------------------------------------------
+//  Copyright Microsoft Corporation
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//  http://www.apache.org/licenses/LICENSE-2.0
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+//  ----------------------------------------------------------------------------------
+
+namespace DurableTask.AzureStorage.Tests
+{
+    using DurableTask.AzureStorage;
+    using DurableTask.Core;
+    using Microsoft.VisualStudio.TestTools.UnitTesting;
+    using System;
+    using System.Runtime.Serialization;
+    using System.Threading.Tasks;
+
+    [TestClass]
+    public class QueueClientEncodingStrategyIntegrationTests
+    {
+        private const string TestConnectionString = "UseDevelopmentStorage=true";
+
+        [TestMethod]
+        // Verifies that messages sent with Base64 encoding can be processed by a worker with UTF8 encoding
+        public async Task CrossEncodingCompatibility_Base64ToUtf8()
+        {
+            string testName = "Base64ToUtf8";
+            string input = "世界! 🌍 Test with émojis and spéciål chàracters: ñáéíóúü";
+
+            // Create service with Base64 encoding to send messages
+            var base64Settings = new AzureStorageOrchestrationServiceSettings
+            {
+                TaskHubName = testName,
+                StorageAccountClientProvider = new StorageAccountClientProvider(TestConnectionString),
+                QueueClientEncodingStrategy = QueueClientEncodingStrategy.Base64,
+            };
+
+            var base64Service = new AzureStorageOrchestrationService(base64Settings);
+            await base64Service.CreateIfNotExistsAsync();
+            // DON'T start the service - this prevents the dequeue loop from running
+
+            try
+            {
+                // Create orchestration instance with Base64 encoding
+                var base64Client = new TaskHubClient(base64Service);
+                var instance = await base64Client.CreateOrchestrationInstanceAsync(typeof(HelloOrchestrator), input);
+
+                // Create worker with UTF8 encoding to process the message
+                var utf8Settings = new AzureStorageOrchestrationServiceSettings
+                {
+                    TaskHubName = testName,
+                    StorageAccountClientProvider = new StorageAccountClientProvider(TestConnectionString),
+                    QueueClientEncodingStrategy = QueueClientEncodingStrategy.None,
+                };
+
+                var utf8Service = new AzureStorageOrchestrationService(utf8Settings);
+                var utf8Client = new TaskHubClient(utf8Service);
+                var worker = new TaskHubWorker(utf8Service);
+                worker.AddTaskOrchestrations(typeof(HelloOrchestrator));
+                worker.AddTaskActivities(typeof(Hello));
+
+                await worker.StartAsync();
+
+                try
+                {
+                    // Wait for the orchestration to complete using UTF8 worker
+                    var state = await utf8Client.WaitForOrchestrationAsync(instance, TimeSpan.FromSeconds(60));
+
+                    // Verify UTF8 worker successfully processed the Base64 encoded message
+                    Assert.IsNotNull(state);
+                    Assert.AreEqual(OrchestrationStatus.Completed, state.OrchestrationStatus);
+                    Assert.AreEqual($"\"Hello, {input}!\"", state.Output);
+                }
+                finally
+                {
+                    await worker.StopAsync();
+                }
+            }
+            finally
+            {
+                await base64Service.DeleteAsync();
+            }
+        }
+
+        [TestMethod]
+        // Verifies that messages sent with UTF8 encoding can be processed by a worker with Base64 encoding
+        public async Task CrossEncodingCompatibility_Utf8ToBase64()
+        {
+            string testName = "Utf8ToBase64test0";
+            string input = "世界! 🌍 Test with émojis and spéciål chàracters: ñáéíóúü";
+
+            // Create service with UTF8 encoding to send messages
+            var utf8Settings = new AzureStorageOrchestrationServiceSettings
+            {
+                TaskHubName = testName,
+                StorageAccountClientProvider = new StorageAccountClientProvider(TestConnectionString),
+                QueueClientEncodingStrategy = QueueClientEncodingStrategy.None,
+            };
+
+            var utf8Service = new AzureStorageOrchestrationService(utf8Settings);
+            await utf8Service.CreateIfNotExistsAsync();
+            // DON'T start the service - this prevents the dequeue loop from running
+
+            try
+            {
+                // Create orchestration instance with UTF8 encoding
+                var utf8Client = new TaskHubClient(utf8Service);
+                var instance = await utf8Client.CreateOrchestrationInstanceAsync(typeof(HelloOrchestrator), input);
+
+                // Create worker with Base64 encoding to process the message
+                var base64Settings = new AzureStorageOrchestrationServiceSettings
+                {
+                    TaskHubName = testName,
+                    StorageAccountClientProvider = new StorageAccountClientProvider(TestConnectionString),
+                    QueueClientEncodingStrategy = QueueClientEncodingStrategy.Base64,
+                };
+
+                var base64Service = new AzureStorageOrchestrationService(base64Settings);
+                var base64Client = new TaskHubClient(base64Service);
+                var worker = new TaskHubWorker(base64Service);
+                worker.AddTaskOrchestrations(typeof(HelloOrchestrator));
+                worker.AddTaskActivities(typeof(Hello));
+
+                await worker.StartAsync();
+
+                try
+                {
+                    // Wait for the orchestration to complete using Base64 worker
+                    var state = await base64Client.WaitForOrchestrationAsync(instance, TimeSpan.FromSeconds(60));
+
+                    // Verify Base64 worker successfully processed the UTF8 encoded message
+                    Assert.IsNotNull(state);
+                    Assert.AreEqual(OrchestrationStatus.Completed, state.OrchestrationStatus);
+                    Assert.AreEqual($"\"Hello, {input}!\"", state.Output);
+                }
+                finally
+                {
+                    await worker.StopAsync();
+                }
+            }
+            finally
+            {
+                await utf8Service.DeleteAsync();
+            }
+        }
+
+        [TestMethod]
+        // Verifies that messages sent with Base64 encoding can be processed by a worker with Base64 encoding
+        public async Task SameEncodingStrategy_Base64ToBase64()
+        {
+            string testName = "Base64ToBase64";
+            string input = "世界! 🌍 Test with émojis and spéciål chàracters: ñáéíóúü";
+
+            var base64Settings = new AzureStorageOrchestrationServiceSettings
+            {
+                TaskHubName = testName,
+                StorageAccountClientProvider = new StorageAccountClientProvider(TestConnectionString),
+                QueueClientEncodingStrategy = QueueClientEncodingStrategy.Base64,
+            };
+
+            var base64Service = new AzureStorageOrchestrationService(base64Settings);
+
+            try
+            {
+                var base64Client = new TaskHubClient(base64Service);
+                var worker = new TaskHubWorker(base64Service);
+                worker.AddTaskOrchestrations(typeof(HelloOrchestrator));
+                worker.AddTaskActivities(typeof(Hello));
+
+                await worker.StartAsync();
+
+                try
+                {
+                    var instance = await base64Client.CreateOrchestrationInstanceAsync(typeof(HelloOrchestrator), input);
+                    var state = await base64Client.WaitForOrchestrationAsync(instance, TimeSpan.FromSeconds(60));
+
+                    // Verify Base64 worker successfully processed Base64 encoded message
+                    Assert.IsNotNull(state);
+                    Assert.AreEqual(OrchestrationStatus.Completed, state.OrchestrationStatus);
+                    Assert.AreEqual($"\"Hello, {input}!\"", state.Output);
+                }
+                finally
+                {
+                    await worker.StopAsync();
+                }
+            }
+            finally
+            {
+                await base64Service.DeleteAsync();
+            }
+        }
+
+        [TestMethod]
+        // Verifies that messages sent with UTF8 encoding can be processed by a worker with UTF8 encoding
+        public async Task SameEncodingStrategy_Utf8ToUtf8()
+        {
+            string testName = "Utf8ToUtf8";
+            string input = "世界! 🌍 Test with émojis and spéciål chàracters: ñáéíóúü";
+
+            var utf8Settings = new AzureStorageOrchestrationServiceSettings
+            {
+                TaskHubName = testName,
+                StorageAccountClientProvider = new StorageAccountClientProvider(TestConnectionString),
+                QueueClientEncodingStrategy = QueueClientEncodingStrategy.None,
+            };
+
+            var utf8Service = new AzureStorageOrchestrationService(utf8Settings);
+
+            try
+            {
+                var utf8Client = new TaskHubClient(utf8Service);
+                var worker = new TaskHubWorker(utf8Service);
+                worker.AddTaskOrchestrations(typeof(HelloOrchestrator));
+                worker.AddTaskActivities(typeof(Hello));
+
+                await worker.StartAsync();
+
+                try
+                {
+                    var instance = await utf8Client.CreateOrchestrationInstanceAsync(typeof(HelloOrchestrator), input);
+                    var state = await utf8Client.WaitForOrchestrationAsync(instance, TimeSpan.FromSeconds(60));
+
+                    // Verify UTF8 worker successfully processed UTF8 encoded message
+                    Assert.IsNotNull(state);
+                    Assert.AreEqual(OrchestrationStatus.Completed, state.OrchestrationStatus);
+                    Assert.AreEqual($"\"Hello, {input}!\"", state.Output);
+                }
+                finally
+                {
+                    await worker.StopAsync();
+                }
+            }
+            finally
+            {
+                await utf8Service.DeleteAsync();
+            }
+        }
+
+        // Test orchestrator and activity
+        [KnownType(typeof(Hello))]
+        internal class HelloOrchestrator : TaskOrchestration<string, string>
+        {
+            public override async Task<string> RunTask(OrchestrationContext context, string input)
+            {
+                // Wait for 5 seconds before executing the activity (shorter for faster tests)
+                // await context.CreateTimer<object>(context.CurrentUtcDateTime.AddSeconds(5), null);
+                return await context.ScheduleTask<string>(typeof(Hello), input);
+            }
+        }
+
+        internal class Hello : TaskActivity<string, string>
+        {
+            protected override string Execute(TaskContext context, string input)
+            {
+                if (string.IsNullOrEmpty(input))
+                {
+                    throw new ArgumentNullException(nameof(input));
+                }
+
+                return $"Hello, {input}!";
+            }
+        }
+    }
+} 

src/DurableTask.AzureStorage/AzureStorageOrchestrationServiceSettings.cs

@@ -288,5 +288,11 @@ internal LogHelper Logger
         /// Consumers that require separate dispatch (such as the new out-of-proc v2 SDKs) must set this to true.
         /// </summary>
         public bool UseSeparateQueueForEntityWorkItems { get; set; } = false;
+
+        /// <summary>
+        /// Gets or sets the encoding strategy used for Azure Storage Queue messages.
+        /// The default is <see cref="QueueClientEncodingStrategy.None"/>.
+        /// </summary>
+        public QueueClientEncodingStrategy QueueClientEncodingStrategy { get; set; } = QueueClientEncodingStrategy.None;
     }
 }

src/DurableTask.AzureStorage/MessageManager.cs

@@ -156,16 +156,71 @@ public async Task<MessageData> DeserializeQueueMessageAsync(QueueMessage queueMe
         {
             // TODO: Deserialize with Stream?
             byte[] body = queueMessage.Body.ToArray();
-            MessageData envelope;
-            try
+            MessageData envelope = null;
+
+            if (this.azureStorageClient.Settings.QueueClientEncodingStrategy == QueueClientEncodingStrategy.Base64)
             {
-                envelope = this.DeserializeMessageData(Encoding.UTF8.GetString(body));
+                try
+                {
+                    // For a queue client using the Base64 encoding strategy, error handling is configured via the options.
+                    // Therefore, we should always receive messages encoded in Base64 here.
+                    string messageText = queueMessage.Body.ToString();
+                    envelope = this.DeserializeMessageData(messageText);
+                }
+                catch (Exception ex)
+                {
+                    this.azureStorageClient.Settings.Logger.GeneralWarning(
+                        this.azureStorageClient.QueueAccountName,
+                        this.azureStorageClient.Settings.TaskHubName,
+                        $"Failed to process message with Base64 client fallback. MessageId: {queueMessage.MessageId}, Error: {ex.Message}");
+
+                        // Re-throw the original exception to maintain the original behavior
+                        throw;
+                }
             }
-            catch(JsonReaderException)
+            else // queue client use Utf8-encoding.
+            {
+                try
+                {
+                    // For UTF8 encoding strategy, try to decode as UTF8 first
+                    envelope = this.DeserializeMessageData(Encoding.UTF8.GetString(body));
+                }
+                catch (JsonReaderException)
+                {
+                    // If UTF8 decode fails, the message might be Base64 encoded
+                    // This can happen when MessageDecodingFailed event was triggered
+                    // In this case, queueMessage.Body contains raw Base64 bytes
+                    try
+                    {
+                        // Convert raw bytes to Base64 string, then decode Base64 to get original message
+                        string base64String = Encoding.UTF8.GetString(body);
+                        byte[] decodedBytes = Convert.FromBase64String(base64String);
+                        string decodedMessage = Encoding.UTF8.GetString(decodedBytes);
+                        envelope = this.DeserializeMessageData(decodedMessage);
+
+                        // Log successful cross-encoding processing
+                        this.azureStorageClient.Settings.Logger.GeneralWarning(
+                            this.azureStorageClient.QueueAccountName,
+                            this.azureStorageClient.Settings.TaskHubName,
+                            $"Successfully processed Base64 message with UTF8 client. MessageId: {queueMessage.MessageId}");
+                    }
+                    catch (Exception fallbackException)
+                    {
+                        // Log the fallback failure
+                        this.azureStorageClient.Settings.Logger.GeneralWarning(
+                            this.azureStorageClient.QueueAccountName,
+                            this.azureStorageClient.Settings.TaskHubName,
+                            $"Failed to process message with UTF8 client fallback. MessageId: {queueMessage.MessageId}, Error: {fallbackException.Message}");
+
+                        // Re-throw the original exception to maintain the original behavior
+                        throw;
+                    }
+                }
+            }
+
+            if (envelope == null)
             {
-                // This catch block is a hotfix and better implementation might be needed in future. 
-                // DTFx.AzureStorage 1.x and 2.x use different encoding methods. Adding this line to enable forward compatibility.
-                envelope = this.DeserializeMessageData(Encoding.UTF8.GetString(Convert.FromBase64String(Encoding.UTF8.GetString(body))));
+                throw new InvalidOperationException($"Failed to deserialize message {queueMessage.MessageId} with encoding strategy {this.azureStorageClient.Settings.QueueClientEncodingStrategy}");
             }
 
             if (!string.IsNullOrEmpty(envelope.CompressedBlobName))

src/DurableTask.AzureStorage/QueueClientEncodingStrategy.cs

@@ -0,0 +1,36 @@
+//  ----------------------------------------------------------------------------------
+//  Copyright Microsoft Corporation
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//  http://www.apache.org/licenses/LICENSE-2.0
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+//  ----------------------------------------------------------------------------------
+
+using Azure.Storage.Queues;
+
+namespace DurableTask.AzureStorage
+{
+    /// <summary>
+    /// Specifies the encoding strategy used for Azure Storage Queue messages.
+    /// This enum maps to the Azure Storage SDK's <see cref="QueueMessageEncoding"/> values.
+    /// </summary>
+    public enum QueueClientEncodingStrategy
+    {
+        /// <summary>
+        /// Use UTF8 encoding for queue messages. Maps to <see cref="QueueMessageEncoding.None"/>.
+        /// </summary>
+        None = 0,
+
+        /// <summary>
+        /// Use Base64 encoding for queue messages. Maps to <see cref="QueueMessageEncoding.Base64"/>.
+        /// This provides compatibility with older versions and may be required in some scenarios 
+        /// where UTF8 encoding is not supported.
+        /// </summary>
+        Base64 = 1,
+    }
+}