Add support for dedupe statuses in isolated

Directory.Packages.props

@@ -27,9 +27,9 @@
     <PackageVersion Include="Microsoft.CodeAnalysis.CSharp" Version="3.9.0" />
     <PackageVersion Include="Microsoft.CodeAnalysis.CSharp.Workspaces" Version="3.9.0" />
     <PackageVersion Include="Microsoft.CodeAnalysis.Workspaces.Common" Version="3.9.0" />
-    <PackageVersion Include="Microsoft.DurableTask.Client.Grpc" Version="1.20.1" />
-    <PackageVersion Include="Microsoft.DurableTask.Worker.Grpc" Version="1.20.1" />
-    <PackageVersion Include="Microsoft.DurableTask.Abstractions" Version="1.20.1" />
+    <PackageVersion Include="Microsoft.DurableTask.Client.Grpc" Version="1.22.0" />
+    <PackageVersion Include="Microsoft.DurableTask.Worker.Grpc" Version="1.22.0" />
+    <PackageVersion Include="Microsoft.DurableTask.Abstractions" Version="1.22.0" />
     <PackageVersion Include="Microsoft.DurableTask.Analyzers" Version="0.2.0" />
     <PackageVersion Include="Microsoft.Extensions.Azure" Version="1.7.0" />
     <PackageVersion Include="Microsoft.Extensions.Caching.Memory" Version="10.0.0" />
@@ -58,8 +58,8 @@
     <PackageVersion Include="Microsoft.Azure.DurableTask.Redis" Version="0.1.9-alpha" />
     <PackageVersion Include="Microsoft.Azure.Functions.Worker" Version="2.2.0" />
     <PackageVersion Include="Microsoft.Azure.Functions.Worker.ApplicationInsights" Version="2.0.0" />
-    <PackageVersion Include="Microsoft.Azure.Functions.Worker.Extensions.DurableTask.AzureManaged" Version="0.4.2-alpha" />
-    <PackageVersion Include="Microsoft.Azure.Functions.Worker.Extensions.DurableTask.SqlServer" Version="1.5.2" />
+    <PackageVersion Include="Microsoft.Azure.Functions.Worker.Extensions.DurableTask.AzureManaged" Version="1.4.0" />
+    <PackageVersion Include="Microsoft.Azure.Functions.Worker.Extensions.DurableTask.SqlServer" Version="1.5.4" />
     <PackageVersion Include="Microsoft.Azure.Functions.Worker.Extensions.Http" Version="3.3.0" />
     <PackageVersion Include="Microsoft.Azure.Functions.Worker.Extensions.Http.AspNetCore" Version="2.0.2" />
     <PackageVersion Include="Microsoft.Azure.Functions.Worker.Sdk" Version="2.0.5"/>

src/WebJobs.Extensions.DurableTask/ContextImplementations/DurableClient.cs

@@ -1140,16 +1140,6 @@ async Task<string> IDurableOrchestrationClient.RestartAsync(string instanceId, b
             // GetOrchestrationInstanceStateAsync will throw ArgumentException if the provided instanceid is not found.
             OrchestrationState state = await this.GetOrchestrationInstanceStateAsync(instanceId);
 
-            bool isInstanceNotCompleted = state.OrchestrationStatus == OrchestrationStatus.Running ||
-                                        state.OrchestrationStatus == OrchestrationStatus.Pending ||
-                                        state.OrchestrationStatus == OrchestrationStatus.Suspended;
-
-            if (isInstanceNotCompleted && !restartWithNewInstanceId)
-            {
-                throw new InvalidOperationException($"Instance '{instanceId}' cannot be restarted while it is in state '{state.OrchestrationStatus}'. " +
-                    "Wait until it has completed, or restart with a new instance ID.");
-            }
-
             JToken input = ParseToJToken(state.Input);
 
             string newInstanceId = null;

src/WebJobs.Extensions.DurableTask/DurabilityProvider.cs

@@ -1,12 +1,14 @@
-// Copyright (c) .NET Foundation. All rights reserved.
+// Copyright (c) .NET Foundation. All rights reserved.
 // Licensed under the MIT License. See LICENSE in the project root for license information.
 
 using System;
 using System.Collections.Generic;
+using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using DurableTask.Core;
 using DurableTask.Core.Entities;
+using DurableTask.Core.Exceptions;
 using DurableTask.Core.History;
 using DurableTask.Core.Query;
 using Microsoft.Azure.WebJobs.Host.Scale;
@@ -38,6 +40,7 @@ public class DurabilityProvider :
         private readonly IOrchestrationServiceClient innerServiceClient;
         private readonly IEntityOrchestrationService entityOrchestrationService;
         private readonly string connectionName;
+        private readonly int orchestrationCreationRequestTimeoutInSeconds = 180;
 
         /// <summary>
         /// Creates the default <see cref="DurabilityProvider"/>.
@@ -406,10 +409,63 @@ public Task CreateTaskOrchestrationAsync(TaskMessage creationMessage)
             return this.GetOrchestrationServiceClient().CreateTaskOrchestrationAsync(creationMessage);
         }
 
-        /// <inheritdoc />
-        public Task CreateTaskOrchestrationAsync(TaskMessage creationMessage, OrchestrationStatus[] dedupeStatuses)
-        {
-            return this.GetOrchestrationServiceClient().CreateTaskOrchestrationAsync(creationMessage, dedupeStatuses);
+        /// <summary>
+        /// Creates a new task orchestration instance using the specified creation message and dedupe statuses.
+        /// </summary>
+        /// <param name="creationMessage">The creation message for the orchestration.</param>
+        /// <param name="dedupeStatuses">An array of orchestration statuses used for "dedupping":
+        /// If an orchestration with the same instance ID already exists, and its status is in this array, then a
+        /// <see cref="OrchestrationAlreadyExistsException"/> will be thrown.
+        /// If the array contains all of the running statuses (<see cref="OrchestrationStatus.Pending"/>, <see cref="OrchestrationStatus.Running"/>,
+        /// and <see cref="OrchestrationStatus.Suspended"/>), then only terminal statuses can be reused.
+        /// If at least one of these statuses is not included in the array, then if an instance with that status is found, it will first be terminated
+        /// before a new orchestration is created. If the existing instance does not reach a terminal state within 3 minutes, the operation will be cancelled.
+        /// </param>
+        /// <returns>A task that completes when the creation message for the task orchestration instance is enqueued.</returns>
+        /// <exception cref="OrchestrationAlreadyExistsException">Thrown if an orchestration with the same instance ID already exists and its status
+        /// is in <paramref name="dedupeStatuses"/>.</exception>
+        /// <exception cref="OperationCanceledException">Thrown if an existing running instance does not reach a terminal state within 3 minutes.</exception>
+        /// <exception cref="ArgumentException">
+        /// Thrown if <paramref name="dedupeStatuses"/> contains <see cref="OrchestrationStatus.Terminated"/> but also allows at least one running status
+        /// to be reusable. In this case, an existing orchestration with that running status would be terminated, but the creation of the new orchestration
+        /// would immediately fail due to the existing orchestration now having status <see cref="OrchestrationStatus.Terminated"/>.
+        /// </exception>
+        public async Task CreateTaskOrchestrationAsync(TaskMessage creationMessage, OrchestrationStatus[] dedupeStatuses)
+        {
+            using var timeoutCts = new CancellationTokenSource(TimeSpan.FromSeconds(this.orchestrationCreationRequestTimeoutInSeconds));
+            await this.CreateTaskOrchestrationAsync(creationMessage, dedupeStatuses, timeoutCts.Token);
+        }
+
+        /// <summary>
+        /// Creates a new task orchestration instance using the specified creation message and dedupe statuses.
+        /// </summary>
+        /// <param name="creationMessage">The creation message for the orchestration.</param>
+        /// <param name="dedupeStatuses">An array of orchestration statuses used for "dedupping":
+        /// If an orchestration with the same instance ID already exists, and its status is in this array, then a
+        /// <see cref="OrchestrationAlreadyExistsException"/> will be thrown.
+        /// If the array contains all of the running statuses (<see cref="OrchestrationStatus.Pending"/>, <see cref="OrchestrationStatus.Running"/>,
+        /// and <see cref="OrchestrationStatus.Suspended"/>), then only terminal statuses can be reused.
+        /// If at least one of these statuses is not included in the array, then if an instance with that status is found, it will first be terminated
+        /// before a new orchestration is created. This method will wait for the instance to reach a terminal status for a maximum of one hour or
+        /// until the <paramref name="cancellationToken"/> is invoked, whichever occurs first.</param>
+        /// <param name="cancellationToken">The cancellation token used to cancel waiting for an existing instance to terminate in the case that a
+        /// non-terminal instance is found whose runtime status is not included in <paramref name="dedupeStatuses"/>.</param>
+        /// <returns>A task that completes when the creation message for the task orchestration instance is enqueued.</returns>
+        /// <exception cref="OrchestrationAlreadyExistsException">Thrown if an orchestration with the same instance ID already exists and its status
+        /// is in <paramref name="dedupeStatuses"/>.</exception>
+        /// <exception cref="OperationCanceledException">Thrown if the operation is cancelled via <paramref name="cancellationToken"/>.</exception>
+        /// <exception cref="ArgumentException">
+        /// Thrown if <paramref name="dedupeStatuses"/> contains <see cref="OrchestrationStatus.Terminated"/> but also allows at least one running status
+        /// to be reusable. In this case, an existing orchestration with that running status would be terminated, but the creation of the new orchestration
+        /// would immediately fail due to the existing orchestration now having status <see cref="OrchestrationStatus.Terminated"/>.
+        /// </exception>
+        public async virtual Task CreateTaskOrchestrationAsync(TaskMessage creationMessage, OrchestrationStatus[] dedupeStatuses, CancellationToken cancellationToken)
+        {
+            await this.TerminateTaskOrchestrationWithReusableRunningStatusAndWaitAsync(
+                creationMessage.OrchestrationInstance.InstanceId,
+                dedupeStatuses,
+                cancellationToken);
+            await this.GetOrchestrationServiceClient().CreateTaskOrchestrationAsync(creationMessage, dedupeStatuses);
         }
 
         /// <inheritdoc />
@@ -614,10 +670,90 @@ public virtual Task<IAsyncEnumerable<HistoryEvent>> StreamOrchestrationHistoryAs
         /// <summary>
         /// Attempts to modify the durability service's UseSeparateQueueForEntityWorkItems property.
         /// </summary>
-        /// <param name="newValue">The value to set</param>
+        /// <param name="newValue">The value to set.</param>
         public virtual void SetUseSeparateQueueForEntityWorkItems(bool newValue)
         {
             throw this.GetNotImplementedException(nameof(this.SetUseSeparateQueueForEntityWorkItems));
         }
+
+        /// <summary>
+        /// If an orchestration exists with a status that is not in <paramref name="dedupeStatuses"/> and has a running status (one of
+        /// <see cref="OrchestrationStatus.Pending"/>, <see cref="OrchestrationStatus.Running"/>, or <see cref="OrchestrationStatus.Suspended"/>),
+        /// then this method terminates the specified orchestration instance and waits until:
+        /// - The orchestration's status changes to <see cref="OrchestrationStatus.Terminated"/>,
+        /// - or the orchestration is deleted,
+        /// - or the operation is cancelled via the <paramref name="cancellationToken"/>.
+        /// </summary>
+        /// <param name="instanceId">The instance ID of the orchestration.</param>
+        /// <param name="dedupeStatuses">The dedupe statuses of the orchestration.</param>
+        /// <param name="cancellationToken">The cancellation token.</param>
+        /// <returns>A task that completes when any of the above conditions are reached.</returns>
+        /// <exception cref="OperationCanceledException">Thrown if the operation is cancelled via the <paramref name="cancellationToken"/>.</exception>
+        /// <exception cref="OrchestrationAlreadyExistsException">Thrown if an orchestration already exists with status in <paramref name="dedupeStatuses"/>.</exception>
+        /// <exception cref="ArgumentException">Thrown if <paramref name="dedupeStatuses"/> contains <see cref="OrchestrationStatus.Terminated"/> but allows
+        /// at least one running status to be reusable.</exception>
+        private async Task TerminateTaskOrchestrationWithReusableRunningStatusAndWaitAsync(
+            string instanceId,
+            OrchestrationStatus[] dedupeStatuses,
+            CancellationToken cancellationToken)
+        {
+            var runningStatuses = new List<OrchestrationStatus>()
+            {
+                OrchestrationStatus.Running,
+                OrchestrationStatus.Pending,
+                OrchestrationStatus.Suspended,
+            };
+
+            if (dedupeStatuses != null && runningStatuses.Any(
+                status => !dedupeStatuses.Contains(status)) && dedupeStatuses.Contains(OrchestrationStatus.Terminated))
+            {
+                throw new ArgumentException(
+                    "Invalid dedupe statuses: cannot include 'Terminated' while also allowing reuse of running instances, " +
+                    "because the running instance would be terminated and then immediately conflict with the dedupe check.");
+            }
+
+            bool IsRunning(OrchestrationStatus status) =>
+                runningStatuses.Contains(status);
+
+            // At least one running status is reusable, so determine if an orchestration already exists with this status and terminate it if so
+            if (dedupeStatuses == null || runningStatuses.Any(status => !dedupeStatuses.Contains(status)))
+            {
+                OrchestrationState orchestrationState = await this.GetOrchestrationStateAsync(instanceId, executionId: null);
+
+                if (orchestrationState != null)
+                {
+                    if (dedupeStatuses?.Contains(orchestrationState.OrchestrationStatus) == true)
+                    {
+                        throw new OrchestrationAlreadyExistsException($"An orchestration with instance ID '{instanceId}' and status " +
+                            $"'{orchestrationState.OrchestrationStatus}' already exists");
+                    }
+
+                    if (IsRunning(orchestrationState.OrchestrationStatus))
+                    {
+                        // Check for cancellation before attempting to terminate the orchestration
+                        cancellationToken.ThrowIfCancellationRequested();
+
+                        string dedupeStatusesDescription = dedupeStatuses == null
+                            ? "null (all statuses reusable)"
+                            : dedupeStatuses.Length == 0
+                                ? "[] (all statuses reusable)"
+                                : $"[{string.Join(", ", dedupeStatuses)}]";
+
+                        string terminationReason = $"A new instance creation request has been issued for instance {instanceId} which " +
+                            $"currently has status {orchestrationState.OrchestrationStatus}. Since the dedupe statuses of the creation request, " +
+                            $"{dedupeStatusesDescription}, do not contain the orchestration's status, the orchestration has been terminated " +
+                            $"and a new instance with the same instance ID will be created.";
+
+                        await this.ForceTerminateTaskOrchestrationAsync(instanceId, terminationReason);
+
+                        await this.WaitForOrchestrationAsync(
+                            instanceId,
+                            orchestrationState.OrchestrationInstance.ExecutionId,
+                            TimeSpan.FromHours(1),
+                            cancellationToken);
+                    }
+                }
+            }
+        }
     }
-}
+}

src/WebJobs.Extensions.DurableTask/DurableTaskExtension.cs

@@ -1480,7 +1480,7 @@ Task<HttpResponseMessage> IAsyncConverter<HttpRequestMessage, HttpResponseMessag
             HttpRequestMessage request,
             CancellationToken cancellationToken)
         {
-            return this.HttpApiHandler.HandleRequestAsync(request);
+            return this.HttpApiHandler.HandleRequestAsync(request, cancellationToken);
         }
 
         internal static string ValidatePayloadSize(string payload)

src/WebJobs.Extensions.DurableTask/HttpApiHandler.cs

@@ -12,6 +12,7 @@
 using System.Threading;
 using System.Threading.Tasks;
 using DurableTask.Core;
+using DurableTask.Core.Exceptions;
 using DurableTask.Core.History;
 using Microsoft.AspNetCore.Routing;
 using Microsoft.AspNetCore.Routing.Template;
@@ -285,7 +286,7 @@ internal async Task<HttpResponseMessage> WaitForCompletionOrCreateCheckStatusRes
             }
         }
 
-        public async Task<HttpResponseMessage> HandleRequestAsync(HttpRequestMessage request)
+        public async Task<HttpResponseMessage> HandleRequestAsync(HttpRequestMessage request, CancellationToken cancellationToken)
         {
             try
             {
@@ -309,7 +310,7 @@ public async Task<HttpResponseMessage> HandleRequestAsync(HttpRequestMessage req
                     string instanceId = (string)routeValues[InstanceIdRouteParameter];
                     if (request.Method == HttpMethod.Post)
                     {
-                        return await this.HandleStartOrchestratorRequestAsync(request, functionName, instanceId);
+                        return await this.HandleStartOrchestratorRequestAsync(request, functionName, instanceId, cancellationToken);
                     }
                     else
                     {
@@ -906,7 +907,8 @@ private async Task<HttpResponseMessage> HandleResumeInstanceRequestAsync(
         private async Task<HttpResponseMessage> HandleStartOrchestratorRequestAsync(
             HttpRequestMessage request,
             string functionName,
-            string instanceId)
+            string instanceId,
+            CancellationToken cancellationToken)
         {
             try
             {
@@ -965,7 +967,8 @@ await durableClient.DurabilityProvider.CreateTaskOrchestrationAsync(
                             Event = executionStartedEvent,
                             OrchestrationInstance = instance,
                         },
-                        this.config.Options.OverridableExistingInstanceStates.ToDedupeStatuses());
+                        this.config.Options.OverridableExistingInstanceStates.ToDedupeStatuses(),
+                        cancellationToken);
                 }
                 else
                 {
@@ -994,6 +997,14 @@ await durableClient.DurabilityProvider.CreateTaskOrchestrationAsync(
             {
                 return request.CreateErrorResponse(HttpStatusCode.BadRequest, "Invalid JSON content", e);
             }
+            catch (OrchestrationAlreadyExistsException e)
+            {
+                return request.CreateErrorResponse(HttpStatusCode.Conflict, e.Message);
+            }
+            catch (ArgumentException e)
+            {
+                return request.CreateErrorResponse(HttpStatusCode.BadRequest, e.Message);
+            }
         }
 
         private static string GetHeaderValueFromHeaders(string header, HttpRequestHeaders headers)
@@ -1068,6 +1079,10 @@ private async Task<HttpResponseMessage> HandleRestartInstanceRequestAsync(
             {
                 return request.CreateErrorResponse(HttpStatusCode.BadRequest, "InstanceId does not match a valid orchestration instance.", e);
             }
+            catch (OrchestrationAlreadyExistsException e)
+            {
+                return request.CreateErrorResponse(HttpStatusCode.Conflict, "A non-terminal instance with this instance ID already exists.", e);
+            }
             catch (JsonReaderException e)
             {
                 return request.CreateErrorResponse(HttpStatusCode.BadRequest, "Invalid JSON content", e);