Add RetryPolicy.Handle Property to Allow for Exception Filtering on Retries

src/Abstractions/RetryPolicy.cs

@@ -20,6 +20,7 @@ public class RetryPolicy
     /// The maximum time to delay between attempts, regardless of<paramref name="backoffCoefficient"/>.
     /// </param>
     /// <param name="retryTimeout">The overall timeout for retries.</param>
+    /// <param name="handle">Delegate to call on exception to determine if retries should proceed.</param>
     /// <remarks>
     /// The value <see cref="Timeout.InfiniteTimeSpan"/> can be used to specify an unlimited timeout for
     /// <paramref name="maxRetryInterval"/> or <paramref name="retryTimeout"/>.
@@ -39,7 +40,8 @@ public RetryPolicy(
         TimeSpan firstRetryInterval,
         double backoffCoefficient = 1.0,
         TimeSpan? maxRetryInterval = null,
-        TimeSpan? retryTimeout = null)
+        TimeSpan? retryTimeout = null,
+        Func<TaskFailedException, bool>? handle = null)
     {
         if (maxNumberOfAttempts <= 0)
         {
@@ -86,6 +88,7 @@ public RetryPolicy(
         this.BackoffCoefficient = backoffCoefficient;
         this.MaxRetryInterval = maxRetryInterval ?? TimeSpan.FromHours(1);
         this.RetryTimeout = retryTimeout ?? Timeout.InfiniteTimeSpan;
+        this.SetHandler(handle);
     }
 
     /// <summary>
@@ -123,6 +126,50 @@ public RetryPolicy(
     /// </value>
     public TimeSpan RetryTimeout { get; }
 
+    /// <summary>
+    /// Gets a delegate to call on exception to determine if retries should proceed.
+    /// </summary>
+    /// <value>
+    /// Defaults delegate that always returns true (i.e., all exceptions are retried).
+    /// </value>
+    public Func<Exception, bool> Handle { get; private set; }
+
+    /// <summary>
+    /// Set <see cref="Handle"/> delegate property.
+    /// </summary>
+    /// <param name="handle">
+    /// Deletegate that receives <see cref="TaskFailedException"/> and returns boolean that
+    /// determines if the task should be retried.
+    /// </param>
+    /// <exception cref="InvalidOperationException">
+    /// This represents a defect in this library in that it should always receive wither
+    /// <see cref="global::DurableTask.Core.Exceptions.TaskFailedException"/> or
+    /// <see cref="global::DurableTask.Core.Exceptions.SubOrchestrationFailedException"/>.
+    /// </exception>
+    void SetHandler(Func<TaskFailedException, bool>? handle)
+    {
+        this.Handle = handle is null
+            ? ((ex) => true)
+            : ((ex) =>
+            {
+                if (ex is global::DurableTask.Core.Exceptions.TaskFailedException globalTaskFailedException)
+                {
+                    var taskFailedException = new TaskFailedException(globalTaskFailedException.Name, globalTaskFailedException.ScheduleId, globalTaskFailedException);
+                    return handle.Invoke(taskFailedException);
+                }
+                else if (ex is global::DurableTask.Core.Exceptions.SubOrchestrationFailedException globalSubOrchestrationFailedException)
+                {
+                    var taskFailedException = new TaskFailedException(globalSubOrchestrationFailedException.Name, globalSubOrchestrationFailedException.ScheduleId, globalSubOrchestrationFailedException);
+                    return handle.Invoke(taskFailedException);
+                }
+                else
+                {
+                    throw new InvalidOperationException("TaskFailedException nor SubOrchestrationFailedException were not received.");
+                }
+            });
+    }
+
+
 #pragma warning disable SA1623 // Property summary documentation should match accessors
     /// <summary>
     /// This functionality is not implemented. Will be removed in the future. Use TaskOptions.FromRetryHandler instead.

src/Shared/Core/RetryPolicyExtensions.cs

@@ -28,6 +28,7 @@ static TimeSpan ConvertInfiniteTimeSpans(TimeSpan timeout) =>
             BackoffCoefficient = retry.BackoffCoefficient,
             MaxRetryInterval = ConvertInfiniteTimeSpans(retry.MaxRetryInterval),
             RetryTimeout = ConvertInfiniteTimeSpans(retry.RetryTimeout),
+            Handle = retry.Handle,
         };
     }
 }

test/Grpc.IntegrationTests/OrchestrationErrorHandling.cs

@@ -222,6 +222,66 @@ public async Task RetryActivityFailuresCustomLogic(int expectedNumberOfAttempts,
         Assert.Equal(expectedNumberOfAttempts, actualNumberOfAttempts);
     }
 
+    [Theory]
+    [InlineData(10, typeof(ApplicationException), false, int.MaxValue, 2, 1, OrchestrationRuntimeStatus.Failed)] // 1 attempt since retry timeout expired.
+    [InlineData(2, typeof(ApplicationException), false, int.MaxValue, null, 1, OrchestrationRuntimeStatus.Failed)] // 1 attempt since handler specifies no retry.
+    [InlineData(2, typeof(CustomException),true, int.MaxValue, null, 2, OrchestrationRuntimeStatus.Failed)] // 2 attempts, custom exception type
+    [InlineData(10, typeof(XunitException),true, 4, null, 5, OrchestrationRuntimeStatus.Completed)] // 10 attempts, 3rd party exception type
+    public async Task RetryActivityFailuresCustomLogicAndPolicy(
+        int maxNumberOfAttempts,
+        Type exceptionType,
+        bool retryException,
+        int exceptionCount,
+        int? retryTimeout,
+        int expectedNumberOfAttempts,
+        OrchestrationRuntimeStatus expRuntimeStatus)
+    {
+        string errorMessage = "Kah-BOOOOOM!!!"; // Use an obviously fake error message to avoid confusion when debugging
+
+        int actualNumberOfAttempts = 0;
+        int retryHandlerCalls = 0;
+        RetryPolicy retryPolicy = new(
+            maxNumberOfAttempts,
+            firstRetryInterval: TimeSpan.FromMilliseconds(1),
+            backoffCoefficient: 2,
+            retryTimeout: retryTimeout.HasValue ? TimeSpan.FromMilliseconds(retryTimeout.Value) : null,
+            handle: taskFailedException =>
+            {
+                retryHandlerCalls++;
+                return taskFailedException.FailureDetails.IsCausedBy(exceptionType) && retryException;
+            });
+        TaskOptions taskOptions = TaskOptions.FromRetryPolicy(retryPolicy);
+
+
+        TaskName orchestratorName = "BustedOrchestration";
+        await using HostTestLifetime server = await this.StartWorkerAsync(b =>
+        {
+            b.AddTasks(tasks =>
+                tasks.AddOrchestratorFunc(orchestratorName, async ctx =>
+                {
+                    await ctx.CallActivityAsync("Foo", options: taskOptions);
+                })
+                .AddActivityFunc("Foo", (TaskActivityContext context) =>
+                {
+                    if (actualNumberOfAttempts++ < exceptionCount)
+                    {
+                        throw MakeException(exceptionType, errorMessage);
+                    }
+                }));
+        });
+
+        string instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync(orchestratorName);
+        OrchestrationMetadata metadata = await server.Client.WaitForInstanceCompletionAsync(
+            instanceId, getInputsAndOutputs: true, this.TimeoutToken);
+
+        Assert.NotNull(metadata);
+        Assert.Equal(instanceId, metadata.InstanceId);
+        Assert.Equal(expRuntimeStatus, metadata.RuntimeStatus);
+        // More calls to retry handler than expected.
+        //Assert.Equal(expectedNumberOfAttempts, retryHandlerCalls);
+        Assert.Equal(expectedNumberOfAttempts, actualNumberOfAttempts);
+    }
+
     /// <summary>
     /// Tests retry policies for sub-orchestration calls.
     /// </summary>
@@ -269,6 +329,76 @@ public async Task RetrySubOrchestrationFailures(int expectedNumberOfAttempts, Ty
         Assert.True(metadata.FailureDetails.IsCausedBy<TaskFailedException>());
     }
 
+    [Theory]
+    [InlineData(10, typeof(ApplicationException), false, int.MaxValue, 2, 1, OrchestrationRuntimeStatus.Failed)] // 1 attempt since retry timeout expired.
+    [InlineData(2, typeof(ApplicationException), false, int.MaxValue, null, 1, OrchestrationRuntimeStatus.Failed)] // 1 attempt since handler specifies no retry.
+    [InlineData(2, typeof(CustomException), true, int.MaxValue, null, 2, OrchestrationRuntimeStatus.Failed)] // 2 attempts, custom exception type
+    [InlineData(10, typeof(XunitException), true, 4, null, 5, OrchestrationRuntimeStatus.Completed)] // 10 attempts, 3rd party exception type
+    public async Task RetrySubOrchestratorFailuresCustomLogicAndPolicy(
+        int maxNumberOfAttempts,
+        Type exceptionType,
+        bool retryException,
+        int exceptionCount,
+        int? retryTimeout,
+        int expectedNumberOfAttempts,
+        OrchestrationRuntimeStatus expRuntimeStatus)
+    {
+        string errorMessage = "Kah-BOOOOOM!!!"; // Use an obviously fake error message to avoid confusion when debugging
+
+        int actualNumberOfAttempts = 0;
+        int retryHandlerCalls = 0;
+        RetryPolicy retryPolicy = new(
+            maxNumberOfAttempts,
+            firstRetryInterval: TimeSpan.FromMilliseconds(1),
+            backoffCoefficient: 2,
+            retryTimeout: retryTimeout.HasValue ? TimeSpan.FromMilliseconds(retryTimeout.Value) : null,
+            handle: taskFailedException =>
+            {
+                retryHandlerCalls++;
+                return taskFailedException.FailureDetails.IsCausedBy(exceptionType) && retryException;
+            });
+        TaskOptions taskOptions = TaskOptions.FromRetryPolicy(retryPolicy);
+
+        TaskName orchestratorName = "OrchestrationWithBustedSubOrchestrator";
+        await using HostTestLifetime server = await this.StartWorkerAsync(b =>
+        {
+            b.AddTasks(tasks =>
+                tasks.AddOrchestratorFunc(orchestratorName, async ctx =>
+                {
+                    await ctx.CallSubOrchestratorAsync("BustedSubOrchestrator", options: taskOptions);
+                })
+                .AddOrchestratorFunc("BustedSubOrchestrator", context =>
+                {
+                    if (actualNumberOfAttempts++ < exceptionCount)
+                    {
+                        throw MakeException(exceptionType, errorMessage);
+                    }
+                }));
+        });
+
+        string instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync(orchestratorName);
+        OrchestrationMetadata metadata = await server.Client.WaitForInstanceCompletionAsync(
+            instanceId, getInputsAndOutputs: true, this.TimeoutToken);
+
+        Assert.NotNull(metadata);
+        Assert.Equal(instanceId, metadata.InstanceId);
+        Assert.Equal(expRuntimeStatus, metadata.RuntimeStatus);
+        // More calls to retry handler than expected.
+        //Assert.Equal(expectedNumberOfAttempts, retryHandlerCalls);
+        Assert.Equal(expectedNumberOfAttempts, actualNumberOfAttempts);
+
+        // The root orchestration failed due to a failure with the sub-orchestration, resulting in a TaskFailedException
+        if (expRuntimeStatus == OrchestrationRuntimeStatus.Failed)
+        {
+            Assert.NotNull(metadata.FailureDetails);
+            Assert.True(metadata.FailureDetails!.IsCausedBy<TaskFailedException>());
+        }
+        else
+        {
+            Assert.Null(metadata.FailureDetails);
+        }
+    }
+
     [Theory]
     [InlineData(1, typeof(ApplicationException))] // 1 attempt, built-in exception type
     [InlineData(2, typeof(CustomException))] // 2 attempts, custom exception type