[Durable] Implement retry policy support (#590)

Author: AnatoliB

.gitignore

@@ -63,8 +63,10 @@ PowerShell.sln.DotSettings.user
 StyleCop.Cache
 
 examples/PSCoreApp/Modules
-src/Modules
-!src/Modules/Microsoft.Azure.Functions.PowerShellWorker
+src/Modules/Microsoft.PowerShell.*
+src/Modules/PackageManagement
+src/Modules/PowerShellGet
+src/Modules/ThreadJob
 
 # protobuf
 protobuf/*

examples/durable/DurableApp/FlakyActivity/function.json

@@ -0,0 +1,9 @@
+{
+  "bindings": [
+    {
+      "name": "name",
+      "type": "activityTrigger",
+      "direction": "in"
+    }
+  ]
+}

examples/durable/DurableApp/FlakyActivity/run.ps1

@@ -0,0 +1,9 @@
+param($name)
+
+# Intentional intermittent error
+$random = Get-Random -Minimum 0.0 -Maximum 1.0
+if ($random -gt 0.2) {
+    throw 'Nope, no luck this time...'
+}
+
+"Hello $name"

examples/durable/DurableApp/FunctionChainingWithRetriesOrchestrator/function.json

@@ -0,0 +1,9 @@
+{
+  "bindings": [
+    {
+      "name": "Context",
+      "type": "orchestrationTrigger",
+      "direction": "in"
+    }
+  ]
+}

examples/durable/DurableApp/FunctionChainingWithRetriesOrchestrator/run.ps1

@@ -0,0 +1,17 @@
+using namespace System.Net
+
+param($Context)
+
+$ErrorActionPreference = 'Stop'
+
+$output = @()
+
+$retryOptions = New-DurableRetryOptions `
+                    -FirstRetryInterval (New-Timespan -Seconds 1) `
+                    -MaxNumberOfAttempts 7
+
+$output += Invoke-ActivityFunction -FunctionName 'FlakyActivity' -Input 'Tokyo' -RetryOptions $retryOptions
+$output += Invoke-ActivityFunction -FunctionName 'FlakyActivity' -Input 'Seattle' -RetryOptions $retryOptions
+$output += Invoke-ActivityFunction -FunctionName 'FlakyActivity' -Input 'London' -RetryOptions $retryOptions
+
+$output

examples/durable/DurableApp/FunctionChainingWithRetriesStart/function.json

@@ -0,0 +1,24 @@
+{
+  "bindings": [
+    {
+      "authLevel": "function",
+      "type": "httpTrigger",
+      "direction": "in",
+      "name": "Request",
+      "methods": [
+        "get",
+        "post"
+      ]
+    },
+    {
+      "type": "http",
+      "direction": "out",
+      "name": "Response"
+    },
+    {
+      "name": "starter",
+      "type": "durableClient",
+      "direction": "in"
+    }    
+  ]
+}

examples/durable/DurableApp/FunctionChainingWithRetriesStart/run.ps1

@@ -0,0 +1,13 @@
+using namespace System.Net
+
+param($Request, $TriggerMetadata)
+
+Write-Host 'FunctionChainingWithRetriesStart started'
+
+$InstanceId = Start-NewOrchestration -FunctionName 'FunctionChainingWithRetriesOrchestrator' -InputObject 'Hello'
+Write-Host "Started orchestration with ID = '$InstanceId'"
+
+$Response = New-OrchestrationCheckStatusResponse -Request $Request -InstanceId $InstanceId
+Push-OutputBinding -Name Response -Value $Response
+
+Write-Host 'FunctionChainingWithRetriesStart completed'

src/Durable/Actions/CallActivityWithRetryAction.cs

@@ -0,0 +1,71 @@
+//
+// Copyright (c) Microsoft. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+//
+
+using System;
+using System.Collections.Generic;
+
+namespace Microsoft.Azure.Functions.PowerShellWorker.Durable.Actions
+{
+    /// <summary>
+    /// An orchestration action that represents calling an activity function with retry.
+    /// </summary>
+    internal class CallActivityWithRetryAction : OrchestrationAction
+    {
+        /// <summary>
+        /// The activity function name.
+        /// </summary>
+        public readonly string FunctionName;
+
+        /// <summary>
+        /// The input to the activity function.
+        /// </summary>
+        public readonly object Input;
+
+        /// <summary>
+        /// Retry options.
+        /// </summary>
+        public readonly Dictionary<string, object> RetryOptions;
+
+        public CallActivityWithRetryAction(string functionName, object input, RetryOptions retryOptions)
+            : base(ActionType.CallActivityWithRetry)
+        {
+            FunctionName = functionName;
+            Input = input;
+            RetryOptions = ToDictionary(retryOptions);
+        }
+
+        private static Dictionary<string, object> ToDictionary(RetryOptions retryOptions)
+        {
+            var result = new Dictionary<string, object>()
+                            {
+                                { "firstRetryIntervalInMilliseconds", ToIntMilliseconds(retryOptions.FirstRetryInterval) },
+                                { "maxNumberOfAttempts", retryOptions.MaxNumberOfAttempts }
+                            };
+
+            AddOptionalValue(result, "backoffCoefficient", retryOptions.BackoffCoefficient, x => x);
+            AddOptionalValue(result, "maxRetryIntervalInMilliseconds", retryOptions.MaxRetryInterval, ToIntMilliseconds);
+            AddOptionalValue(result, "retryTimeoutInMilliseconds", retryOptions.RetryTimeout, ToIntMilliseconds);
+
+            return result;
+        }
+
+        private static void AddOptionalValue<T>(
+            Dictionary<string, object> dictionary,
+            string name,
+            T? nullable,
+            Func<T, object> transformValue) where T : struct
+        {
+            if (nullable.HasValue)
+            {
+                dictionary.Add(name, transformValue(nullable.Value));
+            }
+        }
+
+        private static object ToIntMilliseconds(TimeSpan timespan)
+        {
+            return (int)timespan.TotalMilliseconds;
+        }
+    }
+}

src/Durable/Commands/InvokeActivityFunctionCommand.cs

@@ -33,6 +33,10 @@ public class InvokeActivityFunctionCommand : PSCmdlet
         [Parameter]
         public SwitchParameter NoWait { get; set; }
 
+        [Parameter]
+        [ValidateNotNull]
+        public RetryOptions RetryOptions { get; set; }
+
         private readonly DurableTaskHandler _durableTaskHandler = new DurableTaskHandler();
 
         protected override void EndProcessing()
@@ -41,11 +45,14 @@ protected override void EndProcessing()
             var context = (OrchestrationContext)privateData[SetFunctionInvocationContextCommand.ContextKey];
             var loadedFunctions = FunctionLoader.GetLoadedFunctions();
 
-            var task = new ActivityInvocationTask(FunctionName, Input);
+            var task = new ActivityInvocationTask(FunctionName, Input, RetryOptions);
             ActivityInvocationTask.ValidateTask(task, loadedFunctions);
 
             _durableTaskHandler.StopAndInitiateDurableTaskOrReplay(
-                task, context, NoWait.IsPresent, WriteObject, failureReason => DurableActivityErrorHandler.Handle(this, failureReason));
+                task, context, NoWait.IsPresent,
+                output: WriteObject,
+                onFailure: failureReason => DurableActivityErrorHandler.Handle(this, failureReason),
+                retryOptions: RetryOptions);
         }
 
         protected override void StopProcessing()

src/Durable/DurableTaskHandler.cs

@@ -20,7 +20,8 @@ public void StopAndInitiateDurableTaskOrReplay(
             OrchestrationContext context,
             bool noWait,
             Action<object> output,
-            Action<string> onFailure)
+            Action<string> onFailure,
+            RetryOptions retryOptions = null)
         {
             context.OrchestrationActionCollector.Add(task.CreateOrchestrationAction());
 
@@ -30,6 +31,8 @@ public void StopAndInitiateDurableTaskOrReplay(
             }
             else
             {
+                context.OrchestrationActionCollector.NextBatch();
+
                 var scheduledHistoryEvent = task.GetScheduledHistoryEvent(context);
                 var completedHistoryEvent = task.GetCompletedHistoryEvent(context, scheduledHistoryEvent);
 
@@ -56,7 +59,32 @@ public void StopAndInitiateDurableTaskOrReplay(
                             break;
 
                         case HistoryEventType.TaskFailed:
-                            onFailure(completedHistoryEvent.Reason);
+                            if (retryOptions == null)
+                            {
+                                onFailure(completedHistoryEvent.Reason);
+                            }
+                            else
+                            {
+                                // Reset IsProcessed, let RetryProcessor handle these events instead.
+                                scheduledHistoryEvent.IsProcessed = false;
+                                completedHistoryEvent.IsProcessed = false;
+
+                                var shouldContinueProcessing =
+                                    RetryProcessor.Process(
+                                        context.History,
+                                        scheduledHistoryEvent,
+                                        retryOptions.MaxNumberOfAttempts,
+                                        onSuccess:
+                                            result => {
+                                                output(TypeExtensions.ConvertFromJson(result));
+                                            },
+                                        onFailure);
+                                        
+                                if (shouldContinueProcessing)
+                                {
+                                    InitiateAndWaitForStop(context);
+                                }
+                            }
                             break;
                     }
                 }
@@ -73,6 +101,8 @@ public void WaitAll(
             OrchestrationContext context,
             Action<object> output)
         {
+            context.OrchestrationActionCollector.NextBatch();
+                
             var completedEvents = new List<HistoryEvent>();
             foreach (var task in tasksToWaitFor)
             {
@@ -118,6 +148,8 @@ public void WaitAny(
             OrchestrationContext context,
             Action<object> output)
         {
+            context.OrchestrationActionCollector.NextBatch();
+                
             var completedTasks = new List<DurableTask>();
             DurableTask firstCompletedTask = null;
             int firstCompletedHistoryEventIndex = -1;