Added support for external events and the ability to cancel an awaited durable timer (#527)

Co-authored-by: Michael Peng <[email protected]>

Author: michaelpeng36

examples/durable/DurableApp/HumanInteractionOrchestrator/run.ps1

@@ -6,21 +6,23 @@ Write-Host 'HumanInteractionOrchestrator: started.'
 
 $output = @()
 
-$duration = $Context.Input.Duration
+$duration = New-TimeSpan -Seconds $Context.Input.Duration
+$managerId = $Context.Input.ManagerId
+$skipManagerId = $Context.Input.SkipManagerId
 
-Invoke-ActivityFunction -FunctionName "RequestApproval"
+$output += Invoke-ActivityFunction -FunctionName "RequestApproval" -Input $managerId
 
 $durableTimeoutEvent = Start-DurableTimer -Duration $duration -NoWait
 $approvalEvent = Start-DurableExternalEventListener -EventName "ApprovalEvent" -NoWait
 
 $firstEvent = Wait-DurableTask -Task @($approvalEvent, $durableTimeoutEvent) -Any
 
 if ($approvalEvent -eq $firstEvent) {
-    Stop-DurableTimerTask -TimerTask $durableTimeout
+    Stop-DurableTimerTask -Task $durableTimeoutEvent
     $output += Invoke-ActivityFunction -FunctionName "ProcessApproval" -Input $approvalEvent
 }
 else {
-    $output += Invoke-ActivityFunction -FunctionName "EscalateApproval"
+    $output += Invoke-ActivityFunction -FunctionName "EscalateApproval" -Input $skipManagerId
 }
 
 Write-Host 'HumanInteractionOrchestrator: finished.'

examples/durable/DurableApp/HumanInteractionStart/run.ps1

@@ -4,7 +4,7 @@ param($Request, $TriggerMetadata)
 
 Write-Host 'HumanInteractionStart started'
 
-$OrchestratorInputs = @{ Duration = 10 }
+$OrchestratorInputs = @{ Duration = 45; ManagerId = 1; SkipManagerId = 2 }
 
 $InstanceId = Start-NewOrchestration -FunctionName 'HumanInteractionOrchestrator' -InputObject $OrchestratorInputs
 Write-Host "Started orchestration with ID = '$InstanceId'"

examples/durable/DurableApp/RequestApproval/run.ps1

@@ -1,3 +1,3 @@
 param($name)
 
-Write-Host "Approval requested."
+"Approval requested."

examples/durable/DurableApp/local.settings.json

@@ -3,6 +3,7 @@
   "Values": {
     "AzureWebJobsStorage": "UseDevelopmentStorage=true",
     "FUNCTIONS_WORKER_RUNTIME": "powershell",
+    "FUNCTIONS_WORKER_RUNTIME_VERSION": 7,
     "PSWorkerInProcConcurrencyUpperBound": 10
   }
 }

src/Durable/ActionType.cs

@@ -44,5 +44,15 @@ internal enum ActionType
         /// Wait for an external event.
         /// </summary>
         WaitForExternalEvent = 6,
+
+        /// <summary>
+        /// Call an entity function.
+        /// </summary>
+        CallEntity = 7,
+
+        /// <summary>
+        /// Make an Http call.
+        /// </summary>
+        CallHttp = 8,
     }
 }

src/Durable/ActivityInvocationTask.cs

@@ -17,12 +17,12 @@ public class ActivityInvocationTask : DurableTask
     {
         internal string FunctionName { get; }
 
-        private object FunctionInput { get; }
+        private object Input { get; }
 
         internal ActivityInvocationTask(string functionName, object functionInput)
         {
             FunctionName = functionName;
-            FunctionInput = functionInput;
+            Input = functionInput;
         }
 
         internal override HistoryEvent GetScheduledHistoryEvent(OrchestrationContext context)
@@ -44,7 +44,7 @@ internal override HistoryEvent GetCompletedHistoryEvent(OrchestrationContext con
 
         internal override OrchestrationAction CreateOrchestrationAction()
         {
-            return new CallActivityAction(FunctionName, FunctionInput);
+            return new CallActivityAction(FunctionName, Input);
         }
 
         internal static void ValidateTask(ActivityInvocationTask task, IEnumerable<AzFunctionInfo> loadedFunctions)

src/Durable/CreateDurableTimerAction.cs

@@ -16,11 +16,17 @@ internal class CreateDurableTimerAction : OrchestrationAction
         /// The DateTime at which the timer will fire.
         /// </summary>
         public readonly DateTime FireAt;
-        
-        public CreateDurableTimerAction (DateTime fireAt)
+
+        /// <summary>
+        /// Indicates whether the timer has been canceled.
+        /// </summary>
+        public bool IsCanceled;
+
+        public CreateDurableTimerAction (DateTime fireAt, bool isCanceled = false)
             : base(ActionType.CreateTimer)
         {
             FireAt = fireAt;
+            IsCanceled = isCanceled;
         }
     }
 }

src/Durable/DurableTaskHandler.cs

@@ -31,23 +31,26 @@ public void StopAndInitiateDurableTaskOrReplay(
                 var scheduledHistoryEvent = task.GetScheduledHistoryEvent(context);
                 var completedHistoryEvent = task.GetCompletedHistoryEvent(context, scheduledHistoryEvent);
 
-                // Assume that the task scheduled must have completed if NoWait is not present and the orchestrator restarted
-                if (scheduledHistoryEvent == null)
-                {
-                    InitiateAndWaitForStop(context);
-                }
-
+                // We must check if the task has been completed first, otherwise external events will always wait upon replays
                 if (completedHistoryEvent != null)
                 {                         
                     CurrentUtcDateTimeUpdater.UpdateCurrentUtcDateTime(context);
-
-                    scheduledHistoryEvent.IsProcessed = true;
-                    completedHistoryEvent.IsProcessed = true;
 
                     if (GetEventResult(completedHistoryEvent) != null)
                     {
                         output(GetEventResult(completedHistoryEvent));
                     }
+
+                    if (scheduledHistoryEvent != null)
+                    {
+                        scheduledHistoryEvent.IsProcessed = true;
+                    }
+
+                    completedHistoryEvent.IsProcessed = true;
+                }
+                else if (scheduledHistoryEvent == null)
+                {
+                    InitiateAndWaitForStop(context);
                 }
             }
         }
@@ -69,7 +72,11 @@ public void WaitAll(
                     break;
                 }
 
-                scheduledHistoryEvent.IsProcessed = true;
+                if (scheduledHistoryEvent != null)
+                {
+                    scheduledHistoryEvent.IsProcessed = true;
+                }
+
                 completedHistoryEvent.IsProcessed = true;
                 completedEvents.Add(completedHistoryEvent);
             }
@@ -108,11 +115,13 @@ public void WaitAny(
                 var scheduledHistoryEvent = task.GetScheduledHistoryEvent(context);
                 var completedHistoryEvent = task.GetCompletedHistoryEvent(context, scheduledHistoryEvent);
 
+                // We must mark this event as processed even if it has not completed; subsequent completed history events
+                // corresponding to an awaited task will not have their IsProcessed value ever set to true.
                 if (scheduledHistoryEvent != null)
                 {
                     scheduledHistoryEvent.IsProcessed = true;
                 }
-                
+
                 if (completedHistoryEvent != null)
                 {
                     completedTasks.Add(task);
@@ -149,10 +158,17 @@ public void Stop()
 
         private static object GetEventResult(HistoryEvent historyEvent)
         {
-            // Output the result if and only if the history event is a completed activity function
-            return historyEvent.EventType != HistoryEventType.TaskCompleted
-                ? null
-                : TypeExtensions.ConvertFromJson(historyEvent.Result);
+            // Output the result if and only if the history event is a completed activity function or a raised external event
+            
+            if (historyEvent.EventType == HistoryEventType.TaskCompleted)
+            {
+                return TypeExtensions.ConvertFromJson(historyEvent.Result);
+            }
+            else if (historyEvent.EventType == HistoryEventType.EventRaised)
+            {
+                return TypeExtensions.ConvertFromJson(historyEvent.Input);
+            }
+            return null;
         }
 
         private void InitiateAndWaitForStop(OrchestrationContext context)

src/Durable/DurableTimerTask.cs

@@ -14,13 +14,16 @@ namespace Microsoft.Azure.Functions.PowerShellWorker.Durable
     // All DurableTimerTasks must be complete or canceled for the orchestration to complete
     public class DurableTimerTask : DurableTask
     {
-        public DateTime FireAt { get; set; }
+        internal DateTime FireAt { get; }
+
+        private CreateDurableTimerAction Action { get; }
 
         // Only incomplete, uncanceled DurableTimerTasks should be created
         internal DurableTimerTask(
             DateTime fireAt)
         {
             FireAt = fireAt;
+            Action = new CreateDurableTimerAction(FireAt);
         }
 
         internal override HistoryEvent GetScheduledHistoryEvent(OrchestrationContext context)
@@ -42,7 +45,13 @@ internal override HistoryEvent GetCompletedHistoryEvent(OrchestrationContext con
 
         internal override OrchestrationAction CreateOrchestrationAction()
         {
-            return new CreateDurableTimerAction(FireAt);
+            return Action;
+        }
+
+        // Indicates that the task has been canceled; without this, the orchestration will not terminate until the timer has expired
+        internal void Cancel()
+        {
+            Action.IsCanceled = true;
         }
     }
 }

src/Durable/ExternalEventAction.cs

@@ -0,0 +1,29 @@
+//
+// Copyright (c) Microsoft. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+//
+
+namespace Microsoft.Azure.Functions.PowerShellWorker.Durable
+{
+    /// <summary>
+    /// An orchestration action that represents listening for an external event
+    /// </summary>
+    internal class ExternalEventAction : OrchestrationAction
+    {
+        /// <summary>
+        /// The external event name.
+        /// </summary>
+        public readonly string ExternalEventName;
+
+        /// <summary>
+        /// Reason for the action. This field is necessary for the Durable extension to recognize the ExternalEventAction.
+        /// </summary>
+        public readonly string Reason = "ExternalEvent";
+
+        public ExternalEventAction(string externalEventName)
+            : base(ActionType.WaitForExternalEvent)
+        {
+            ExternalEventName = externalEventName; 
+        }
+    }
+}