Proper workflow retry support in Dapr SDK (#1090)

Signed-off-by: Chris Gillum <[email protected]>
Co-authored-by: halspang <[email protected]>

examples/Workflow/WorkflowConsoleApp/Activities/ReserveInventoryActivity.cs

@@ -1,5 +1,4 @@
-using System.Threading.Tasks;
-using Dapr.Client;
+using Dapr.Client;
 using Dapr.Workflow;
 using Microsoft.Extensions.Logging;
 using WorkflowConsoleApp.Models;
@@ -27,7 +26,7 @@ public override async Task<InventoryResult> RunAsync(WorkflowActivityContext con
                 req.ItemName);
 
             // Ensure that the store has items
-            InventoryItem item = await client.GetStateAsync<InventoryItem>(
+            InventoryItem item = await this.client.GetStateAsync<InventoryItem>(
                 storeName,
                 req.ItemName.ToLowerInvariant());
 

examples/Workflow/WorkflowConsoleApp/Activities/UpdateInventoryActivity.cs

@@ -1,5 +1,4 @@
-using System.Threading.Tasks;
-using Dapr.Client;
+using Dapr.Client;
 using Dapr.Workflow;
 using WorkflowConsoleApp.Models;
 using Microsoft.Extensions.Logging;

examples/Workflow/WorkflowConsoleApp/Workflows/OrderProcessingWorkflow.cs

@@ -6,11 +6,18 @@ namespace WorkflowConsoleApp.Workflows
 {
     public class OrderProcessingWorkflow : Workflow<OrderPayload, OrderResult>
     {
+        readonly WorkflowTaskOptions defaultActivityRetryOptions = new WorkflowTaskOptions
+        {
+            // NOTE: Beware that changing the number of retries is a breaking change for existing workflows.
+            RetryPolicy = new WorkflowRetryPolicy(
+                maxNumberOfAttempts: 3,
+                firstRetryInterval: TimeSpan.FromSeconds(5)),
+        };
+
         public override async Task<OrderResult> RunAsync(WorkflowContext context, OrderPayload order)
         {
             string orderId = context.InstanceId;
 
-
             // Notify the user that an order has come through
             await context.CallActivityAsync(
                 nameof(NotifyActivity),
@@ -19,7 +26,8 @@ await context.CallActivityAsync(
             // Determine if there is enough of the item available for purchase by checking the inventory
             InventoryResult result = await context.CallActivityAsync<InventoryResult>(
                 nameof(ReserveInventoryActivity),
-                new InventoryRequest(RequestId: orderId, order.Name, order.Quantity));
+                new InventoryRequest(RequestId: orderId, order.Name, order.Quantity),
+                this.defaultActivityRetryOptions);
 
             // If there is insufficient inventory, fail and let the user know 
             if (!result.Success)
@@ -34,14 +42,16 @@ await context.CallActivityAsync(
             // There is enough inventory available so the user can purchase the item(s). Process their payment
             await context.CallActivityAsync(
                 nameof(ProcessPaymentActivity),
-                new PaymentRequest(RequestId: orderId, order.Name, order.Quantity, order.TotalCost));
+                new PaymentRequest(RequestId: orderId, order.Name, order.Quantity, order.TotalCost),
+                this.defaultActivityRetryOptions);
 
             try
             {
                 // There is enough inventory available so the user can purchase the item(s). Process their payment
                 await context.CallActivityAsync(
                     nameof(UpdateInventoryActivity),
-                    new PaymentRequest(RequestId: orderId, order.Name, order.Quantity, order.TotalCost));                
+                    new PaymentRequest(RequestId: orderId, order.Name, order.Quantity, order.TotalCost),
+                    this.defaultActivityRetryOptions);                
             }
             catch (WorkflowTaskFailedException e)
             {

examples/Workflow/WorkflowUnitTest/OrderProcessingTests.cs

@@ -1,6 +1,5 @@
 using System.Threading.Tasks;
 using Dapr.Workflow;
-using Microsoft.DurableTask;
 using Moq;
 using WorkflowConsoleApp.Activities;
 using WorkflowConsoleApp.Models;
@@ -24,7 +23,7 @@ public async Task TestSuccessfulOrder()
             // Mock the call to ReserveInventoryActivity
             Mock<WorkflowContext> mockContext = new();
             mockContext
-                .Setup(ctx => ctx.CallActivityAsync<InventoryResult>(nameof(ReserveInventoryActivity), It.IsAny<InventoryRequest>(), It.IsAny<TaskOptions>()))
+                .Setup(ctx => ctx.CallActivityAsync<InventoryResult>(nameof(ReserveInventoryActivity), It.IsAny<InventoryRequest>(), It.IsAny<WorkflowTaskOptions>()))
                 .Returns(Task.FromResult(inventoryResult));
 
             // Run the workflow directly
@@ -36,17 +35,17 @@ public async Task TestSuccessfulOrder()
 
             // Verify that ReserveInventoryActivity was called with a specific input
             mockContext.Verify(
-                ctx => ctx.CallActivityAsync<InventoryResult>(nameof(ReserveInventoryActivity), expectedInventoryRequest, It.IsAny<TaskOptions>()),
+                ctx => ctx.CallActivityAsync<InventoryResult>(nameof(ReserveInventoryActivity), expectedInventoryRequest, It.IsAny<WorkflowTaskOptions>()),
                 Times.Once());
 
             // Verify that ProcessPaymentActivity was called with a specific input
             mockContext.Verify(
-                ctx => ctx.CallActivityAsync(nameof(ProcessPaymentActivity), expectedPaymentRequest, It.IsAny<TaskOptions>()),
+                ctx => ctx.CallActivityAsync(nameof(ProcessPaymentActivity), expectedPaymentRequest, It.IsAny<WorkflowTaskOptions>()),
                 Times.Once());
 
             // Verify that there were two calls to NotifyActivity
             mockContext.Verify(
-                ctx => ctx.CallActivityAsync(nameof(NotifyActivity), It.IsAny<Notification>(), It.IsAny<TaskOptions>()),
+                ctx => ctx.CallActivityAsync(nameof(NotifyActivity), It.IsAny<Notification>(), It.IsAny<WorkflowTaskOptions>()),
                 Times.Exactly(2));
         }
 
@@ -61,25 +60,25 @@ public async Task TestInsufficientInventory()
             // Mock the call to ReserveInventoryActivity
             Mock<WorkflowContext> mockContext = new();
             mockContext
-                .Setup(ctx => ctx.CallActivityAsync<InventoryResult>(nameof(ReserveInventoryActivity), It.IsAny<InventoryRequest>(), It.IsAny<TaskOptions>()))
+                .Setup(ctx => ctx.CallActivityAsync<InventoryResult>(nameof(ReserveInventoryActivity), It.IsAny<InventoryRequest>(), It.IsAny<WorkflowTaskOptions>()))
                 .Returns(Task.FromResult(inventoryResult));
 
             // Run the workflow directly
             OrderResult result = await new OrderProcessingWorkflow().RunAsync(mockContext.Object, order);
 
             // Verify that ReserveInventoryActivity was called with a specific input
             mockContext.Verify(
-                ctx => ctx.CallActivityAsync<InventoryResult>(nameof(ReserveInventoryActivity), expectedInventoryRequest, It.IsAny<TaskOptions>()),
+                ctx => ctx.CallActivityAsync<InventoryResult>(nameof(ReserveInventoryActivity), expectedInventoryRequest, It.IsAny<WorkflowTaskOptions>()),
                 Times.Once());
 
             // Verify that ProcessPaymentActivity was never called
             mockContext.Verify(
-                ctx => ctx.CallActivityAsync(nameof(ProcessPaymentActivity), It.IsAny<PaymentRequest>(), It.IsAny<TaskOptions>()),
+                ctx => ctx.CallActivityAsync(nameof(ProcessPaymentActivity), It.IsAny<PaymentRequest>(), It.IsAny<WorkflowTaskOptions>()),
                 Times.Never());
 
             // Verify that there were two calls to NotifyActivity
             mockContext.Verify(
-                ctx => ctx.CallActivityAsync(nameof(NotifyActivity), It.IsAny<Notification>(), It.IsAny<TaskOptions>()),
+                ctx => ctx.CallActivityAsync(nameof(NotifyActivity), It.IsAny<Notification>(), It.IsAny<WorkflowTaskOptions>()),
                 Times.Exactly(2));
         }
     }

src/Dapr.Workflow/DaprWorkflowContext.cs

@@ -35,14 +35,14 @@ internal DaprWorkflowContext(TaskOrchestrationContext innerContext)
 
         public override bool IsReplaying => this.innerContext.IsReplaying;
 
-        public override Task CallActivityAsync(string name, object? input = null, TaskOptions? options = null)
+        public override Task CallActivityAsync(string name, object? input = null, WorkflowTaskOptions? options = null)
         {
-            return WrapExceptions(this.innerContext.CallActivityAsync(name, input, options));
+            return WrapExceptions(this.innerContext.CallActivityAsync(name, input, options?.ToDurableTaskOptions()));
         }
 
-        public override Task<T> CallActivityAsync<T>(string name, object? input = null, TaskOptions? options = null)
+        public override Task<T> CallActivityAsync<T>(string name, object? input = null, WorkflowTaskOptions? options = null)
         {
-            return WrapExceptions(this.innerContext.CallActivityAsync<T>(name, input, options));
+            return WrapExceptions(this.innerContext.CallActivityAsync<T>(name, input, options?.ToDurableTaskOptions()));
         }
 
         public override Task CreateTimer(TimeSpan delay, CancellationToken cancellationToken = default)
@@ -75,14 +75,14 @@ public override void SetCustomStatus(object? customStatus)
             this.innerContext.SetCustomStatus(customStatus);
         }
 
-        public override Task<TResult> CallChildWorkflowAsync<TResult>(string workflowName, object? input = null, TaskOptions? options = null)
+        public override Task<TResult> CallChildWorkflowAsync<TResult>(string workflowName, object? input = null, ChildWorkflowTaskOptions? options = null)
         {
-            return WrapExceptions(this.innerContext.CallSubOrchestratorAsync<TResult>(workflowName, input, options));
+            return WrapExceptions(this.innerContext.CallSubOrchestratorAsync<TResult>(workflowName, input, options?.ToDurableTaskOptions()));
         }
 
-        public override Task CallChildWorkflowAsync(string workflowName, object? input = null, TaskOptions? options = null)
+        public override Task CallChildWorkflowAsync(string workflowName, object? input = null, ChildWorkflowTaskOptions? options = null)
         {
-            return WrapExceptions(this.innerContext.CallSubOrchestratorAsync(workflowName, input, options));
+            return WrapExceptions(this.innerContext.CallSubOrchestratorAsync(workflowName, input, options?.ToDurableTaskOptions()));
         }
 
         public override void ContinueAsNew(object? newInput = null, bool preserveUnprocessedEvents = true)

src/Dapr.Workflow/WorkflowContext.cs

@@ -16,7 +16,6 @@ namespace Dapr.Workflow
     using System;
     using System.Threading;
     using System.Threading.Tasks;
-    using Microsoft.DurableTask;
 
     /// <summary>
     /// Context object used by workflow implementations to perform actions such as scheduling activities, durable timers, waiting for
@@ -101,7 +100,7 @@ public abstract class WorkflowContext
         /// The activity failed with an unhandled exception. The details of the failure can be found in the
         /// <see cref="WorkflowTaskFailedException.FailureDetails"/> property.
         /// </exception>
-        public virtual Task CallActivityAsync(string name, object? input = null, TaskOptions? options = null)
+        public virtual Task CallActivityAsync(string name, object? input = null, WorkflowTaskOptions? options = null)
         {
             return this.CallActivityAsync<object>(name, input, options);
         }
@@ -110,7 +109,7 @@ public virtual Task CallActivityAsync(string name, object? input = null, TaskOpt
         /// A task that completes when the activity completes or fails. The result of the task is the activity's return value.
         /// </returns>
         /// <inheritdoc cref="CallActivityAsync"/>
-        public abstract Task<T> CallActivityAsync<T>(string name, object? input = null, TaskOptions? options = null);
+        public abstract Task<T> CallActivityAsync<T>(string name, object? input = null, WorkflowTaskOptions? options = null);
 
         /// <summary>
         /// Creates a durable timer that expires after the specified delay.
@@ -212,8 +211,11 @@ public virtual Task CreateTimer(TimeSpan delay, CancellationToken cancellationTo
         /// <typeparam name="TResult">
         /// The type into which to deserialize the child workflow's output.
         /// </typeparam>
-        /// <inheritdoc cref="CallChildWorkflowAsync(string, object?, TaskOptions?)"/>
-        public abstract Task<TResult> CallChildWorkflowAsync<TResult>(string workflowName, object? input = null, TaskOptions? options = null);
+        /// <inheritdoc cref="CallChildWorkflowAsync(string, object?, ChildWorkflowTaskOptions?)"/>
+        public abstract Task<TResult> CallChildWorkflowAsync<TResult>(
+            string workflowName,
+            object? input = null,
+            ChildWorkflowTaskOptions? options = null);
 
         /// <summary>
         /// Executes the specified workflow as a child workflow.
@@ -222,7 +224,8 @@ public virtual Task CreateTimer(TimeSpan delay, CancellationToken cancellationTo
         /// <para>
         /// In addition to activities, workflows can schedule other workflows as <i>child workflows</i>.
         /// A child workflow has its own instance ID, history, and status that is independent of the parent workflow
-        /// that started it.
+        /// that started it. You can use <see cref="ChildWorkflowTaskOptions.InstanceId" /> to specify an instance ID
+        /// for the child workflow. Otherwise, the instance ID will be randomly generated.
         /// </para><para>
         /// Child workflows have many benefits:
         /// <list type="bullet">
@@ -237,15 +240,14 @@ public virtual Task CreateTimer(TimeSpan delay, CancellationToken cancellationTo
         /// exception. Child workflows also support automatic retry policies.
         /// </para><para>
         /// Because child workflows are independent of their parents, terminating a parent workflow does not affect
-        /// any child workflows. You must terminate each child workflow independently using its instance ID, which is
-        /// specified by supplying <see cref="SubOrchestrationOptions" /> in place of <see cref="TaskOptions" />.
+        /// any child workflows. You must terminate each child workflow independently using its instance ID, which
+        /// is specified by <see cref="ChildWorkflowTaskOptions.InstanceId" />.
         /// </para>
         /// </remarks>
         /// <param name="workflowName">The name of the workflow to call.</param>
         /// <param name="input">The serializable input to pass to the child workflow.</param>
         /// <param name="options">
-        /// Additional options that control the execution and processing of the child workflow. Callers can choose to
-        /// supply the derived type <see cref="SubOrchestrationOptions" />.
+        /// Additional options that control the execution and processing of the child workflow.
         /// </param>
         /// <returns>A task that completes when the child workflow completes or fails.</returns>
         /// <exception cref="ArgumentException">The specified workflow does not exist.</exception>
@@ -256,7 +258,10 @@ public virtual Task CreateTimer(TimeSpan delay, CancellationToken cancellationTo
         /// The child workflow failed with an unhandled exception. The details of the failure can be found in the
         /// <see cref="WorkflowTaskFailedException.FailureDetails"/> property.
         /// </exception>
-        public virtual Task CallChildWorkflowAsync(string workflowName, object? input = null, TaskOptions? options = null)
+        public virtual Task CallChildWorkflowAsync(
+            string workflowName,
+            object? input = null,
+            ChildWorkflowTaskOptions? options = null)
         {
             return this.CallChildWorkflowAsync<object>(workflowName, input, options);
         }