pushing bug fix for not honoring a host restarting an extended session

Author: Sophia Tevosyan

src/Worker/Grpc/GrpcOrchestrationRunner.cs

@@ -170,11 +170,14 @@ public static string LoadAndRun(
                 pair => pair.Key,
                 pair => ProtoUtils.ConvertValueToObject(pair.Value));
 
-        OrchestratorExecutionResult? result = null;
+        OrchestratorExecutionResult? result = null;
+        MemoryCache? extendedSessions = null;
+
+        // If any of the request parameters are malformed, we assume the default - extended sessions are not enabled and the orchestration history is attached
         bool addToExtendedSessions = false;
         bool requiresHistory = false;
+        bool pastEventsIncluded = true;
         double extendedSessionIdleTimeoutInSeconds = 0;
-        MemoryCache? extendedSessions = null;
 
         if (properties.TryGetValue("ExtendedSessionIdleTimeoutInSeconds", out object? extendedSessionIdleTimeoutObj)
             && extendedSessionIdleTimeoutObj is double extendedSessionIdleTimeout
@@ -183,13 +186,21 @@ public static string LoadAndRun(
             extendedSessionIdleTimeoutInSeconds = extendedSessionIdleTimeout;
             extendedSessions = extendedSessionsCache.GetOrInitializeCache(extendedSessionIdleTimeoutInSeconds);
         }
+
+        if (properties.TryGetValue("IncludePastEvents", out object? includePastEventsObj)
+            && includePastEventsObj is bool includePastEvents)
+        {
+            pastEventsIncluded = includePastEvents;
+        }
 
         if (properties.TryGetValue("ExtendedSession", out object? isExtendedSessionObj)
             && isExtendedSessionObj is bool isExtendedSession
             && isExtendedSession
             && extendedSessions != null)
-        {
-            if (extendedSessions.TryGetValue(request.InstanceId, out ExtendedSessionState? extendedSessionState) && extendedSessionState is not null)
+        {
+            // If a history was provided, even if we already have an extended session stored, we always want to evict whatever state is in the cache and replace it with a new extended
+            // session based on the provided history
+            if (!pastEventsIncluded && extendedSessions.TryGetValue(request.InstanceId, out ExtendedSessionState? extendedSessionState) && extendedSessionState is not null)
             {
                 OrchestrationRuntimeState runtimeState = extendedSessionState!.RuntimeState;
                 runtimeState.NewEvents.Clear();
@@ -205,20 +216,17 @@ public static string LoadAndRun(
                 }
             }
             else
-            {
+            {
+                extendedSessions.Remove(request.InstanceId);
                 addToExtendedSessions = true;
             }
         }
 
         if (result == null)
         {
-            // If this is the first orchestration execution, then the past events count will be 0 but includePastEvents will be true (there are just none to include).
-            // Otherwise, there is an orchestration history but DurableTask.Core did not attach it since the extended session is still active on its end, but we have since evicted the
+            // DurableTask.Core did not attach the orchestration history since the extended session is still active on its end, but we have since evicted the
             // session and lost the orchestration history so we cannot replay the orchestration.
-            if (pastEvents.Count == 0
-                && (properties.TryGetValue("IncludePastEvents", out object? pastEventsIncludedObj)
-                && pastEventsIncludedObj is bool pastEventsIncluded
-                && !pastEventsIncluded))
+            if (!pastEventsIncluded)
             {
                 requiresHistory = true;
             }

test/Worker/Core.Tests/GrpcOrchestrationRunnerTests.cs

@@ -346,6 +346,41 @@ public async void Stale_ExtendedSessions_Evicted_Async()
         Assert.True(response.RequiresHistory);
     }
 
+    [Fact]
+    public void PastEventIncludes_Means_ExtendedSession_Evicted()
+    {
+        using var extendedSessions = new ExtendedSessionsCache();
+        int extendedSessionIdleTimeout = 5;
+        var historyEvent = new Protobuf.HistoryEvent
+        {
+            EventId = -1,
+            Timestamp = Timestamp.FromDateTime(DateTime.UtcNow),
+            ExecutionStarted = new Protobuf.ExecutionStartedEvent()
+            {
+                OrchestrationInstance = new Protobuf.OrchestrationInstance
+                {
+                    InstanceId = TestInstanceId,
+                    ExecutionId = TestExecutionId,
+                },
+            }
+        };
+        Protobuf.OrchestratorRequest orchestratorRequest = CreateOrchestratorRequest([historyEvent]);
+        orchestratorRequest.Properties.Add(new MapField<string, Value>() {
+            { "IncludePastEvents", Value.ForBool(true) },
+            { "ExtendedSession", Value.ForBool(true) },
+            { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(extendedSessionIdleTimeout) } });
+        byte[] requestBytes = orchestratorRequest.ToByteArray();
+        string requestString = Convert.ToBase64String(requestBytes);
+        GrpcOrchestrationRunner.LoadAndRun(requestString, new CallSubOrchestrationOrchestrator(), extendedSessions);
+        Assert.True(extendedSessions.IsInitialized());
+        Assert.True(extendedSessions.GetOrInitializeCache(extendedSessionIdleTimeout).TryGetValue(TestInstanceId, out object? extendedSession));
+
+        // Now we will retry the same exact request. If the extended session is not evicted, then the request will fail due to duplicate ExecutionStarted events being detected
+        // If the extended session is evicted because IncludePastEvents is true, then the request will succeed and a new extended session will be stored
+        GrpcOrchestrationRunner.LoadAndRun(requestString, new CallSubOrchestrationOrchestrator(), extendedSessions);
+        Assert.True(extendedSessions.GetOrInitializeCache(extendedSessionIdleTimeout).TryGetValue(TestInstanceId, out extendedSession));
+    }
+
     static Protobuf.OrchestratorRequest CreateOrchestratorRequest(IEnumerable<Protobuf.HistoryEvent> newEvents)
     {
         var orchestratorRequest = new Protobuf.OrchestratorRequest()