Optimize Continue As New And Solve Missed Raised Events

Test/DurableTask.AzureStorage.Tests/AzureStorageScenarioTests.cs

@@ -173,7 +173,7 @@ public async Task PurgeInstanceHistoryForSingleInstanceWithoutLargeMessageBlobs(
                 await host.StartAsync();
                 TestOrchestrationClient client = await host.StartOrchestrationAsync(typeof(Orchestrations.Factorial), 110, instanceId);
                 await client.WaitForCompletionAsync(TimeSpan.FromSeconds(30));
-                
+
                 List<HistoryStateEvent> historyEvents = await client.GetOrchestrationHistoryAsync(instanceId);
                 Assert.IsTrue(historyEvents.Count > 0);
 
@@ -538,17 +538,9 @@ public async Task ActorOrchestration(bool enableExtendedSessions)
 
                 // Perform some operations
                 await client.RaiseEventAsync("operation", "incr");
-
-                // TODO: Sleeping to avoid a race condition where multiple ContinueAsNew messages
-                //       are processed by the same instance at the same time, resulting in a corrupt
-                //       storage failure in DTFx.
-                await Task.Delay(2000);
                 await client.RaiseEventAsync("operation", "incr");
-                await Task.Delay(2000);
                 await client.RaiseEventAsync("operation", "incr");
-                await Task.Delay(2000);
                 await client.RaiseEventAsync("operation", "decr");
-                await Task.Delay(2000);
                 await client.RaiseEventAsync("operation", "incr");
                 await Task.Delay(2000);
 
@@ -607,9 +599,7 @@ public async Task ActorOrchestrationDeleteAllLargeMessageBlobs(bool enableExtend
 
             int blobCount = await this.GetBlobCount("test-largemessages", instanceId);
 
-            // Ideally there would only be three blobs at the end of the test.
-            // TODO: https://github.com/Azure/azure-functions-durable-extension/issues/509
-            Assert.AreEqual(9, blobCount);
+            Assert.AreEqual(3, blobCount);
 
             await client.PurgeInstanceHistoryByTimePeriod(
                 startDateTime,
@@ -647,7 +637,7 @@ private async Task<Tuple<string, TestOrchestrationClient>> ValidateCharacterCoun
 
                 // Need to wait for the instance to start before sending events to it.
                 // TODO: This requirement may not be ideal and should be revisited.
-                OrchestrationState orchestrationState = 
+                OrchestrationState orchestrationState =
                     await client.WaitForStartupAsync(TimeSpan.FromSeconds(10));
 
                 // Perform some operations
@@ -745,7 +735,7 @@ public async Task RewindOrchestrationsFail()
 
                 var client1 = await host.StartOrchestrationAsync(
                     typeof(Orchestrations.FactorialOrchestratorFail),
-                    input: 3, 
+                    input: 3,
                     instanceId: singletonInstanceId1);
 
                 var statusFail = await client1.WaitForCompletionAsync(TimeSpan.FromSeconds(30));
@@ -759,7 +749,7 @@ public async Task RewindOrchestrationsFail()
                 input: "Catherine",
                 instanceId: singletonInstanceId2);
 
-                await client1.RewindAsync("Rewind failed orchestration only"); 
+                await client1.RewindAsync("Rewind failed orchestration only");
 
                 var statusRewind = await client1.WaitForCompletionAsync(TimeSpan.FromSeconds(30));
 
@@ -1289,7 +1279,7 @@ public async Task LargeBinaryByteMessagePayloads(bool enableExtendedSessions)
                 string currentDirectory = Directory.GetCurrentDirectory();
                 string originalFilePath = Path.Combine(currentDirectory, originalFileName);
                 byte[] readBytes = File.ReadAllBytes(originalFilePath);
-                
+
                 var client = await host.StartOrchestrationAsync(typeof(Orchestrations.EchoBytes), readBytes);
                 var status = await client.WaitForCompletionAsync(TimeSpan.FromMinutes(1));
 
@@ -1949,7 +1939,7 @@ public override async Task<int> RunTask(OrchestrationContext context, int curren
                     }
 
                     return currentValue;
-                    
+
                 }
 
                 async Task<string> WaitForOperation()
@@ -2228,7 +2218,7 @@ protected override string Execute(TaskContext context, string input)
                 }
             }
 
-            internal class HelloFailMultipleActivity : TaskActivity<string, string> 
+            internal class HelloFailMultipleActivity : TaskActivity<string, string>
             {
                 public static bool ShouldFail1 = true;
                 public static bool ShouldFail2 = true;

src/DurableTask.AzureStorage/AzureStorageOrchestrationService.cs

@@ -67,7 +67,7 @@ public sealed class AzureStorageOrchestrationService :
         readonly TableEntityConverter tableEntityConverter;
 
         readonly ResettableLazy<Task> taskHubCreator;
-        readonly BlobLeaseManager leaseManager; 
+        readonly BlobLeaseManager leaseManager;
         readonly PartitionManager<BlobLease> partitionManager;
         readonly OrchestrationSessionManager orchestrationSessionManager;
         readonly object hubCreationLock;
@@ -274,6 +274,14 @@ public int MaxConcurrentTaskActivityWorkItems
             get { return this.settings.MaxConcurrentTaskActivityWorkItems; }
         }
 
+        /// <summary>
+        ///  Should we carry over unexecuted raised events to the next iteration of an orchestration on ContinueAsNew
+        /// </summary>
+        public BehaviorOnContinueAsNew EventBehaviourForContinueAsNew
+        {
+            get { return this.settings.EventBehaviourForContinueAsNew; }
+        }
+
         // We always leave the dispatcher counts at one unless we can find a customer workload that requires more.
         /// <inheritdoc />
         public int TaskActivityDispatcherCount { get; } = 1;
@@ -355,9 +363,9 @@ public async Task CreateAsync(bool recreateInstanceStore)
         {
             if (recreateInstanceStore)
             {
-               await DeleteTrackingStore();
+                await DeleteTrackingStore();
 
-               this.taskHubCreator.Reset();
+                this.taskHubCreator.Reset();
             }
 
             await this.taskHubCreator.Value;
@@ -816,7 +824,7 @@ public async Task CompleteTaskOrchestrationWorkItemAsync(
             }
 
             session.StartNewLogicalTraceScope();
-            OrchestrationRuntimeState runtimeState = workItem.OrchestrationRuntimeState;
+            OrchestrationRuntimeState runtimeState = newOrchestrationRuntimeState ?? workItem.OrchestrationRuntimeState;
 
             string instanceId = workItem.InstanceId;
             string executionId = runtimeState.OrchestrationInstance.ExecutionId;
@@ -836,7 +844,7 @@ await this.CommitOutboundQueueMessages(
             // will result in a duplicate replay of the orchestration with no side-effects.
             try
             {
-                session.ETag = await this.trackingStore.UpdateStateAsync(runtimeState, instanceId, executionId, session.ETag);
+                session.ETag = await this.trackingStore.UpdateStateAsync(runtimeState, workItem.OrchestrationRuntimeState, instanceId, executionId, session.ETag);
             }
             catch (Exception e)
             {
@@ -1062,7 +1070,7 @@ public async Task CompleteTaskActivityWorkItemAsync(TaskActivityWorkItem workIte
                 // The context does not exist - possibly because it was already removed.
                 AnalyticsEventSource.Log.AssertFailure(
                     this.storageAccountName,
-                    this.settings.TaskHubName, 
+                    this.settings.TaskHubName,
                     $"Could not find context for work item with ID = {workItem.Id}.",
                     Utils.ExtensionVersion);
                 return;
@@ -1120,7 +1128,7 @@ public async Task AbandonTaskActivityWorkItemAsync(TaskActivityWorkItem workItem
                 // The context does not exist - possibly because it was already removed.
                 AnalyticsEventSource.Log.AssertFailure(
                     this.storageAccountName,
-                    this.settings.TaskHubName, 
+                    this.settings.TaskHubName,
                     $"Could not find context for work item with ID = {workItem.Id}.",
                     Utils.ExtensionVersion);
                 return;
@@ -1442,7 +1450,7 @@ public async Task<OrchestrationState> WaitForOrchestrationAsync(
             while (!cancellationToken.IsCancellationRequested && timeout > TimeSpan.Zero)
             {
                 OrchestrationState state = await this.GetOrchestrationStateAsync(instanceId, executionId);
-                if (state == null || 
+                if (state == null ||
                     state.OrchestrationStatus == OrchestrationStatus.Running ||
                     state.OrchestrationStatus == OrchestrationStatus.Pending ||
                     state.OrchestrationStatus == OrchestrationStatus.ContinuedAsNew)

src/DurableTask.AzureStorage/AzureStorageOrchestrationServiceSettings.cs

@@ -14,6 +14,7 @@
 namespace DurableTask.AzureStorage
 {
     using System;
+    using DurableTask.Core;
     using Microsoft.WindowsAzure.Storage.Queue;
     using Microsoft.WindowsAzure.Storage.Table;
 
@@ -139,5 +140,10 @@ public class AzureStorageOrchestrationServiceSettings
         /// If provided, this is used to connect to Azure Storage
         /// </summary>
         public StorageAccountDetails StorageAccountDetails { get; set; }
+
+        /// <summary>
+        ///  Should we carry over unexecuted raised events to the next iteration of an orchestration on ContinueAsNew
+        /// </summary>
+        public BehaviorOnContinueAsNew EventBehaviourForContinueAsNew { get; set; } = BehaviorOnContinueAsNew.Carryover;
     }
 }

src/DurableTask.AzureStorage/Tracking/AzureTableTrackingStore.cs

@@ -851,16 +851,17 @@ public override Task StartAsync()
 
         /// <inheritdoc />
         public override async Task<string> UpdateStateAsync(
-            OrchestrationRuntimeState runtimeState,
+            OrchestrationRuntimeState newRuntimeState,
+            OrchestrationRuntimeState oldRuntimeState,
             string instanceId,
             string executionId,
             string eTagValue)
         {
             int estimatedBytes = 0;
-            IList<HistoryEvent> newEvents = runtimeState.NewEvents;
-            IList<HistoryEvent> allEvents = runtimeState.Events;
+            IList<HistoryEvent> newEvents = newRuntimeState.NewEvents;
+            IList<HistoryEvent> allEvents = newRuntimeState.Events;
 
-            int episodeNumber = Utils.GetEpisodeNumber(runtimeState);
+            int episodeNumber = Utils.GetEpisodeNumber(newRuntimeState);
 
             var newEventListBuffer = new StringBuilder(4000);
             var historyEventBatch = new TableBatchOperation();
@@ -871,7 +872,7 @@ public override async Task<string> UpdateStateAsync(
             {
                 Properties =
                 {
-                    ["CustomStatus"] = new EntityProperty(runtimeState.Status),
+                    ["CustomStatus"] = new EntityProperty(newRuntimeState.Status),
                     ["ExecutionId"] = new EntityProperty(executionId),
                     ["LastUpdatedTime"] = new EntityProperty(newEvents.Last().Timestamp),
                 }

src/DurableTask.AzureStorage/Tracking/ITrackingStore.cs

@@ -64,11 +64,12 @@ interface ITrackingStore
         /// <summary>
         /// Update State in the Tracking store for a particular orchestration instance and execution base on the new runtime state
         /// </summary>
-        /// <param name="runtimeState">The New RuntimeState</param>
+        /// <param name="newRuntimeState">The New RuntimeState</param>
+        /// <param name="oldRuntimeState">The RuntimeState for an olderExecution</param>
         /// <param name="instanceId">InstanceId for the Orchestration Update</param>
         /// <param name="executionId">ExecutionId for the Orchestration Update</param>
         /// <param name="eTag">The ETag value to use for safe updates</param>
-        Task<string> UpdateStateAsync(OrchestrationRuntimeState runtimeState, string instanceId, string executionId, string eTag);
+        Task<string> UpdateStateAsync(OrchestrationRuntimeState newRuntimeState, OrchestrationRuntimeState oldRuntimeState, string instanceId, string executionId, string eTag);
 
         /// <summary>
         /// Get The Orchestration State for the Latest or All Executions

src/DurableTask.AzureStorage/Tracking/InstanceStoreBackedTrackingStore.cs

@@ -128,7 +128,20 @@ public override Task StartAsync()
         }
 
         /// <inheritdoc />
-        public override async Task<string> UpdateStateAsync(OrchestrationRuntimeState runtimeState, string instanceId, string executionId, string eTag)
+        public override async Task<string> UpdateStateAsync(OrchestrationRuntimeState newRuntimeState, OrchestrationRuntimeState oldRuntimeState, string instanceId, string executionId, string eTag)
+        {
+            //In case there is a runtime state for an older execution/iteration as well that needs to be committed, commit it.
+            //This may be the case if a ContinueAsNew was executed on the orchestration
+            if (newRuntimeState != oldRuntimeState)
+            {
+                eTag = await UpdateStateAsync(oldRuntimeState, instanceId, oldRuntimeState.OrchestrationInstance.ExecutionId, eTag);
+            }
+
+            return await UpdateStateAsync(newRuntimeState, instanceId, executionId, eTag);
+        }
+
+        /// <inheritdoc />
+        private async Task<string> UpdateStateAsync(OrchestrationRuntimeState runtimeState, string instanceId, string executionId, string eTag)
         {
             int oldEventsCount = (runtimeState.Events.Count - runtimeState.NewEvents.Count);
             await instanceStore.WriteEntitiesAsync(runtimeState.NewEvents.Select((x, i) =>

src/DurableTask.AzureStorage/Tracking/TrackingStoreBase.cs

@@ -97,6 +97,6 @@ public virtual Task UpdateStatusForRewindAsync(string instanceId)
         public abstract Task StartAsync();
 
         /// <inheritdoc />
-        public abstract Task<string> UpdateStateAsync(OrchestrationRuntimeState runtimeState, string instanceId, string executionId, string eTag);
+        public abstract Task<string> UpdateStateAsync(OrchestrationRuntimeState newRuntimeState, OrchestrationRuntimeState oldRuntimeState, string instanceId, string executionId, string eTag);
     }
 }

src/DurableTask.Core/Command/OrchestrationCompleteOrchestratorAction.cs

@@ -13,8 +13,16 @@
 
 namespace DurableTask.Core.Command
 {
+    using System.Collections.Generic;
+    using DurableTask.Core.History;
+
     internal class OrchestrationCompleteOrchestratorAction : OrchestratorAction
     {
+        public OrchestrationCompleteOrchestratorAction()
+        {
+            CarryoverEvents = new List<HistoryEvent>();
+        }
+
         public OrchestrationStatus OrchestrationStatus;
 
         public override OrchestratorActionType OrchestratorActionType => OrchestratorActionType.OrchestrationComplete;
@@ -24,5 +32,7 @@ internal class OrchestrationCompleteOrchestratorAction : OrchestratorAction
         public string Details { get; set; }
 
         public string NewVersion { get; set; }
+
+        public IList<HistoryEvent> CarryoverEvents { get; }
     }
 }

src/DurableTask.Core/EventHandlingOnContinueAsNew.cs

@@ -0,0 +1,31 @@
+//  ----------------------------------------------------------------------------------
+//  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.Core
+{
+    /// <summary>
+    /// Specifies Behavior to be followed when dealing with unprocessed EventRaisedEvents when an orchestration continues as new
+    /// </summary>
+    public enum BehaviorOnContinueAsNew
+    {
+        /// <summary>
+        /// All pending EventRaisedEvents will be ignored
+        /// </summary>
+        Ignore,
+
+        /// <summary>
+        /// 
+        /// </summary>
+        Carryover,
+    }
+}

src/DurableTask.Core/IOrchestrationService.cs

@@ -97,6 +97,11 @@ public interface IOrchestrationService
         /// </summary>
         int MaxConcurrentTaskOrchestrationWorkItems { get; }
 
+        /// <summary>
+        ///  Should we carry over unexecuted raised events to the next iteration of an orchestration on ContinueAsNew
+        /// </summary>
+        BehaviorOnContinueAsNew EventBehaviourForContinueAsNew { get; }
+
         /// <summary>
         ///     Wait for the next orchestration work item and return the orchestration work item
         /// </summary>

src/DurableTask.Core/Settings/TaskOrchestrationDispatcherSettings.cs

@@ -28,6 +28,7 @@ public TaskOrchestrationDispatcherSettings()
             DispatcherCount = FrameworkConstants.OrchestrationDefaultDispatcherCount;
             MaxConcurrentOrchestrations = FrameworkConstants.OrchestrationDefaultMaxConcurrentItems;
             CompressOrchestrationState = false;
+            EventBehaviourForContinueAsNew = BehaviorOnContinueAsNew.Carryover;
         }
 
         /// <summary>
@@ -55,6 +56,11 @@ public TaskOrchestrationDispatcherSettings()
         /// </summary>
         public bool CompressOrchestrationState { get; set; }
 
+        /// <summary>
+        ///  Should we carry over unexecuted raised events to the next iteration of an orchestration on ContinueAsNew
+        /// </summary>
+        public BehaviorOnContinueAsNew EventBehaviourForContinueAsNew { get; set; }
+
         internal TaskOrchestrationDispatcherSettings Clone()
         {
             return new TaskOrchestrationDispatcherSettings

src/DurableTask.Core/TaskOrchestrationContext.cs

@@ -35,6 +35,13 @@ internal class TaskOrchestrationContext : OrchestrationContext
         bool executionTerminated;
         int idCounter;
 
+        public bool HasContinueAsNew => continueAsNew != null;
+
+        public void AddEventToNextIteration(HistoryEvent he)
+        {
+            continueAsNew.CarryoverEvents.Add(he);
+        }
+
         public TaskOrchestrationContext(OrchestrationInstance orchestrationInstance, TaskScheduler taskScheduler)
         {
             Utils.UnusedParameter(taskScheduler);
@@ -54,6 +61,7 @@ public TaskOrchestrationContext(OrchestrationInstance orchestrationInstance, Tas
         internal void ClearPendingActions()
         {
             this.orchestratorActionsMap.Clear();
+            continueAsNew = null;
         }
 
         public override async Task<TResult> ScheduleTask<TResult>(string name, string version,