Fix failing tests on the DurableTask.Emulator (#267)

* Fixed Continue as New for the local orchestration service. * Match the logic we are using in the ServiceBus orchestration service for setting the session state.
Azure · Mar 20, 2019 · f91ae79 · f91ae79
1 parent 4b29b7a
commit f91ae79
Show file tree

Hide file tree

Showing 2 changed files with 60 additions and 36 deletions.
diff --git a/src/DurableTask.Emulator/LocalOrchestrationService.cs b/src/DurableTask.Emulator/LocalOrchestrationService.cs
@@ -14,6 +14,7 @@
 namespace DurableTask.Emulator
 {
     using DurableTask.Core;
+    using DurableTask.Core.Common;
     using DurableTask.Core.Exceptions;
     using DurableTask.Core.History;
     using Newtonsoft.Json;
@@ -411,10 +412,22 @@ public Task CompleteTaskOrchestrationWorkItemAsync(
         {
             lock (this.thisLock)
             {
+                byte[] newSessionState;
+
+                if (newOrchestrationRuntimeState == null ||
+                newOrchestrationRuntimeState.ExecutionStartedEvent == null ||
+                newOrchestrationRuntimeState.OrchestrationStatus != OrchestrationStatus.Running)
+                {
+                    newSessionState = null;
+                }
+                else
+                {
+                    newSessionState = SerializeOrchestrationRuntimeState(newOrchestrationRuntimeState);
+                }
+
                 this.orchestratorQueue.CompleteSession(
                     workItem.InstanceId,
-                    newOrchestrationRuntimeState != null ?
-                    SerializeOrchestrationRuntimeState(newOrchestrationRuntimeState) : null,
+                    newSessionState,
                     orchestratorMessages,
                     continuedAsNewMessage
                     );
@@ -439,49 +452,62 @@ public Task CompleteTaskOrchestrationWorkItemAsync(
                     }
                 }
 
+                if (workItem.OrchestrationRuntimeState != newOrchestrationRuntimeState)
+                {
+                    var oldState = Utils.BuildOrchestrationState(workItem.OrchestrationRuntimeState);
+                    CommitState(workItem.OrchestrationRuntimeState, oldState).GetAwaiter().GetResult();
+                }
+
                 if (state != null)
                 {
-                    if (!this.instanceStore.TryGetValue(workItem.InstanceId, out Dictionary<string, OrchestrationState> mapState))
-                    {
-                        mapState = new Dictionary<string, OrchestrationState>();
-                        this.instanceStore[workItem.InstanceId] = mapState;
-                    }
+                    CommitState(newOrchestrationRuntimeState, state).GetAwaiter().GetResult();
+                }
+            }
 
-                    mapState[workItem.OrchestrationRuntimeState.OrchestrationInstance.ExecutionId] = state;
+            return Task.FromResult(0);
+        }
 
-                    // signal any waiters waiting on instanceid_executionid or just the latest instanceid_
+        Task CommitState(OrchestrationRuntimeState runtimeState, OrchestrationState state)
+        {
+            if (!this.instanceStore.TryGetValue(runtimeState.OrchestrationInstance.InstanceId, out Dictionary<string, OrchestrationState> mapState))
+            {
+                mapState = new Dictionary<string, OrchestrationState>();
+                this.instanceStore[runtimeState.OrchestrationInstance.InstanceId] = mapState;
+            }
 
-                    if (state.OrchestrationStatus == OrchestrationStatus.Running
-                        || state.OrchestrationStatus == OrchestrationStatus.Pending)
-                    {
-                        return Task.FromResult(0);
-                    }
+            mapState[runtimeState.OrchestrationInstance.ExecutionId] = state;
 
-                    string key = workItem.OrchestrationRuntimeState.OrchestrationInstance.InstanceId + "_" +
-                        workItem.OrchestrationRuntimeState.OrchestrationInstance.ExecutionId;
+            // signal any waiters waiting on instanceid_executionid or just the latest instanceid_
 
-                    string key1 = workItem.OrchestrationRuntimeState.OrchestrationInstance.InstanceId + "_";
+            if (state.OrchestrationStatus == OrchestrationStatus.Running
+                || state.OrchestrationStatus == OrchestrationStatus.Pending)
+            {
+                return Task.FromResult(0);
+            }
 
-                    var tasks = new List<Task>();
+            string key = runtimeState.OrchestrationInstance.InstanceId + "_" +
+                runtimeState.OrchestrationInstance.ExecutionId;
 
-                    if (this.orchestrationWaiters.TryGetValue(key, out TaskCompletionSource<OrchestrationState> tcs))
-                    {
-                        tasks.Add(Task.Run(() => tcs.TrySetResult(state)));
-                    }
+            string key1 = runtimeState.OrchestrationInstance.InstanceId + "_";
 
-                    // for instance id level waiters, we will not consider ContinueAsNew as a terminal state because
-                    // the high level orchestration is still ongoing
-                    if (state.OrchestrationStatus != OrchestrationStatus.ContinuedAsNew
-                        && this.orchestrationWaiters.TryGetValue(key1, out TaskCompletionSource<OrchestrationState> tcs1))
-                    {
-                        tasks.Add(Task.Run(() => tcs1.TrySetResult(state)));
-                    }
+            var tasks = new List<Task>();
 
-                    if (tasks.Count > 0)
-                    {
-                        Task.WaitAll(tasks.ToArray());
-                    }
-                }
+            if (this.orchestrationWaiters.TryGetValue(key, out TaskCompletionSource<OrchestrationState> tcs))
+            {
+                tasks.Add(Task.Run(() => tcs.TrySetResult(state)));
+            }
+
+            // for instance id level waiters, we will not consider ContinueAsNew as a terminal state because
+            // the high level orchestration is still ongoing
+            if (state.OrchestrationStatus != OrchestrationStatus.ContinuedAsNew
+                && this.orchestrationWaiters.TryGetValue(key1, out TaskCompletionSource<OrchestrationState> tcs1))
+            {
+                tasks.Add(Task.Run(() => tcs1.TrySetResult(state)));
+            }
+
+            if (tasks.Count > 0)
+            {
+                Task.WaitAll(tasks.ToArray());
             }
 
             return Task.FromResult(0);

diff --git a/test/DurableTask.Emulator.Tests/EmulatorFunctionalTests.cs b/test/DurableTask.Emulator.Tests/EmulatorFunctionalTests.cs
@@ -54,7 +54,6 @@ await worker.AddTaskOrchestrations(typeof(SimplestGreetingsOrchestration))
             await worker.StopAsync(true);
         }
 
-        [TestCategory("DisabledInCI")] // https://github.com/Azure/durabletask/issues/255
         [TestMethod]
         public async Task MockRecreateOrchestrationTest()
         {
@@ -143,7 +142,6 @@ await worker.AddTaskOrchestrations(typeof(GreetingsRepeatWaitOrchestration))
             await worker.StopAsync(true);
         }
 
-        [TestCategory("DisabledInCI")] // https://github.com/Azure/durabletask/issues/255
         [TestMethod]
         public async Task MockGenerationTest()
         {