Add support for dedupe statuses in isolated

src/WebJobs.Extensions.DurableTask/DurabilityProvider.cs

@@ -3,10 +3,12 @@
 
 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;
@@ -106,7 +108,13 @@ public DurabilityProvider(string storageProviderName, IOrchestrationService serv
         /// <summary>
         /// Event source name (e.g. DurableTask-AzureStorage).
         /// </summary>
-        public virtual string EventSourceName { get; set; }
+        public virtual string EventSourceName { get; set; }
+
+        /// <summary>
+        /// Gets or sets the amount of time in seconds before a creation request for an orchestration times out.
+        /// Default value is 180 seconds.
+        /// </summary>
+        internal int OrchestrationCreationRequestTimeoutInSeconds { get; set; } = 180;
 
         /// <inheritdoc/>
         public int TaskOrchestrationDispatcherCount => this.GetOrchestrationService().TaskOrchestrationDispatcherCount;
@@ -407,9 +415,45 @@ public Task CreateTaskOrchestrationAsync(TaskMessage creationMessage)
         }
 
         /// <inheritdoc />
-        public Task CreateTaskOrchestrationAsync(TaskMessage creationMessage, OrchestrationStatus[] dedupeStatuses)
-        {
-            return this.GetOrchestrationServiceClient().CreateTaskOrchestrationAsync(creationMessage, dedupeStatuses);
+        /// should this method comment be updated too? to mention terminating existing instances?
+        public async virtual Task CreateTaskOrchestrationAsync(TaskMessage creationMessage, OrchestrationStatus[] dedupeStatuses)
+        {
+            using var timeoutCts = new CancellationTokenSource(TimeSpan.FromSeconds(this.OrchestrationCreationRequestTimeoutInSeconds));
+            await this.TerminateTaskOrchestrationWithReusableRunningStatusAndWaitAsync(
+                creationMessage.OrchestrationInstance.InstanceId,
+                dedupeStatuses,
+                timeoutCts.Token);
+            await 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.</param>
+        /// <param name="cancellationToken">The cancellation token.</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>
+        public async Task CreateTaskOrchestrationAsync(TaskMessage creationMessage, OrchestrationStatus[] dedupeStatuses, CancellationToken cancellationToken)
+        {
+            using var timeoutCts = new CancellationTokenSource(TimeSpan.FromSeconds(this.OrchestrationCreationRequestTimeoutInSeconds));
+            using var linkedCts =
+                CancellationTokenSource.CreateLinkedTokenSource(
+                    cancellationToken,
+                    timeoutCts.Token);
+            await this.TerminateTaskOrchestrationWithReusableRunningStatusAndWaitAsync(
+                creationMessage.OrchestrationInstance.InstanceId,
+                dedupeStatuses,
+                linkedCts.Token);
+            await this.GetOrchestrationServiceClient().CreateTaskOrchestrationAsync(creationMessage, dedupeStatuses);
         }
 
         /// <inheritdoc />
@@ -609,6 +653,72 @@ public virtual bool TryGetTargetScaler(
         public virtual Task<IAsyncEnumerable<HistoryEvent>> StreamOrchestrationHistoryAsync(string instanceId, CancellationToken cancellationToken)
         {
             throw this.GetNotImplementedException(nameof(this.StreamOrchestrationHistoryAsync));
+        }
+
+        /// <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>
+        private async Task TerminateTaskOrchestrationWithReusableRunningStatusAndWaitAsync(
+            string instanceId,
+            OrchestrationStatus[] dedupeStatuses,
+            CancellationToken cancellationToken)
+        {
+            var runningStatuses = new List<OrchestrationStatus>()
+            {
+                OrchestrationStatus.Running,
+                OrchestrationStatus.Pending,
+                OrchestrationStatus.Suspended,
+            };
+
+            // At least one running status is reusable, so determine if an orchestration already exists with this status and terminate it if so
+            if (runningStatuses.Any(status => !dedupeStatuses.Contains(status)))
+            {
+                OrchestrationState orchestrationState = await this.GetOrchestrationStateAsync(instanceId, executionId: null);
+
+                if (orchestrationState != null)
+                {
+                    if (dedupeStatuses.Contains(orchestrationState.OrchestrationStatus))
+                    {
+                        throw new OrchestrationAlreadyExistsException($"An orchestration with instance ID '{instanceId}' and status " +
+                            $"'{orchestrationState.OrchestrationStatus}' already exists");
+                    }
+
+                    if (runningStatuses.Contains(orchestrationState.OrchestrationStatus))
+                    {
+                        // Check for cancellation before attempting to terminate the orchestration
+                        cancellationToken.ThrowIfCancellationRequested();
+
+                        await this.ForceTerminateTaskOrchestrationAsync(
+                            instanceId,
+                            $"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, " +
+                            $"{string.Join(", ", dedupeStatuses)}, do not contain the orchestration's status, the orchestration has been " +
+                            $"terminated and a new instance with the same instance ID will be created.");
+
+                        while (orchestrationState != null && orchestrationState.OrchestrationStatus != OrchestrationStatus.Terminated)
+                        {
+                            cancellationToken.ThrowIfCancellationRequested();
+                            await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken);
+                            orchestrationState = await this.GetOrchestrationStateAsync(instanceId, executionId: null);
+                        }
+
+                        // What should we do here? If dedupe statuses contains terminated, then the creation call afterwards will fail.
+                        // Or should we throw an invalid argument exception if dedupeStatuses contains terminated but also allows for reuse of a running status?
+                        // dedupeStatuses = dedupeStatuses.Except(new List<OrchestrationStatus>() { OrchestrationStatus.Terminated }).ToArray();
+                    }
+                }
+            }
         }
     }
 }

src/WebJobs.Extensions.DurableTask/DurableTaskExtension.cs

@@ -183,6 +183,8 @@ public DurableTaskExtension(
             {
                 this.OutOfProcProtocol = OutOfProcOrchestrationProtocol.OrchestratorShim;
             }
+
+            this.defaultDurabilityProvider.OrchestrationCreationRequestTimeoutInSeconds = this.Options.OrchestrationCreationRequestTimeoutInSeconds;
         }
 
         internal DurableTaskOptions Options { get; }

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;
@@ -968,6 +969,17 @@ await durableClient.DurabilityProvider.CreateTaskOrchestrationAsync(
             {
                 return request.CreateErrorResponse(HttpStatusCode.BadRequest, "Invalid JSON content", e);
             }
+            catch (OrchestrationAlreadyExistsException e)
+            {
+                return request.CreateErrorResponse(HttpStatusCode.Conflict, e.Message);
+            }
+            catch (OperationCanceledException)
+            {
+                return request.CreateErrorResponse(
+                    HttpStatusCode.RequestTimeout,
+                    $"Create instance request exceeded timeout of {this.durableTaskOptions.OrchestrationCreationRequestTimeoutInSeconds} " +
+                    $"seconds for instance ID {instanceId} while waiting for the termination of the existing instance with this instance ID.");
+            }
         }
 
         private static string GetHeaderValueFromHeaders(string header, HttpRequestHeaders headers)

src/WebJobs.Extensions.DurableTask/Options/DurableTaskOptions.cs

@@ -6,6 +6,7 @@
 using System.ComponentModel;
 using System.Net.Http;
 using DurableTask.AzureStorage.Partitioning;
+using DurableTask.Core;
 using DurableTask.Core.Settings;
 using Microsoft.Azure.WebJobs.Extensions.DurableTask.Grpc;
 using Microsoft.Azure.WebJobs.Host;
@@ -270,7 +271,22 @@ public string HubName
         /// Default is 100 seconds.
         /// This settings only applies when .NET 6 or greater is used.
         /// </summary>
-        public TimeSpan? GrpcHttpClientTimeout { get; set; } = TimeSpan.FromSeconds(100);
+        public TimeSpan? GrpcHttpClientTimeout { get; set; } = TimeSpan.FromSeconds(100);
+
+        /// <summary>
+        /// Gets or sets the amount of time in seconds before a creation request for an orchestration times out.
+        /// Default value is 180 seconds.
+        /// </summary>
+        /// <remarks>
+        /// This setting is applicable when <see cref="OverridableExistingInstanceStates"/> is set to <see cref="OverridableStates.AnyState"/>.
+        /// If an orchestration in a non-terminal state already exists with the instance ID passed to the creation request, then this
+        /// orchestration will be terminated before the new orchestration is created. This setting controls how long the extension will wait
+        /// for the orchestration to reach a status of <see cref="OrchestrationStatus.Terminated"/> before failing the creation request.
+        /// </remarks>
+        /// <value>
+        /// The number of seconds before a creation request for an orchestration times out.
+        /// </value>
+        public int OrchestrationCreationRequestTimeoutInSeconds { get; set; } = 180;
 
         /// <summary>
         /// Gets or sets the local gRPC listener mode, controlling what version of gRPC listener is created.
@@ -388,6 +404,13 @@ internal void Validate(INameResolver environmentVariableResolver)
             {
                 throw new InvalidOperationException($"{nameof(this.MaxEntityOperationBatchSize)} must be a positive integer value.");
             }
+
+            if (this.OrchestrationCreationRequestTimeoutInSeconds <= 0 || this.OrchestrationCreationRequestTimeoutInSeconds >= 230)
+            {
+                throw new InvalidOperationException($"{nameof(this.OrchestrationCreationRequestTimeoutInSeconds)} must be a positive integer value less than 230 seconds," +
+                    $"which is the maximum amount of time that an HTTP triggered Function can take to respond to a request." +
+                    $"See https://docs.azure.cn/en-us//azure-functions/functions-scale#timeout");
+            }
         }
 
         internal bool IsDefaultHubName()

src/WebJobs.Extensions.DurableTask/TaskHubGrpcServer.cs

@@ -54,6 +54,28 @@ public override Task<Empty> Hello(Empty request, ServerCallContext context)
         {
             try
             {
+                var allStatuses = new List<OrchestrationStatus>()
+                {
+                    OrchestrationStatus.Running,
+                    OrchestrationStatus.Pending,
+                    OrchestrationStatus.Suspended,
+                    OrchestrationStatus.Completed,
+                    OrchestrationStatus.Failed,
+                    OrchestrationStatus.Terminated,
+                };
+
+                // Not all clients are necessarily configured to set the OrchestrationIdReusePolicy field of the request.
+                // If it is null, we assume that they do not support per-request-dedupe statuses, and default to using just
+                // the OverridableExistingInstanceStates setting instead.
+                List<OrchestrationStatus> reusableStatuses = request.OrchestrationIdReusePolicy is null
+                    ? allStatuses
+                    : request.OrchestrationIdReusePolicy.ReplaceableStatus.Select(status => (OrchestrationStatus)status).ToList();
+
+                OrchestrationStatus[] dedupeStatuses = allStatuses
+                    .Except(reusableStatuses)
+                    .Union(this.extension.Options.OverridableExistingInstanceStates.ToDedupeStatuses())
+                    .ToArray();
+
                 // Create the orchestration instance
                 var instance = new OrchestrationInstance
                 {
@@ -86,7 +108,8 @@ await this.GetDurabilityProvider(context).CreateTaskOrchestrationAsync(
                         Event = executionStartedEvent,
                         OrchestrationInstance = instance,
                     },
-                    this.GetStatusesNotToOverride());
+                    dedupeStatuses,
+                    context.CancellationToken);
 
                 return new P.CreateInstanceResponse
                 {
@@ -100,6 +123,15 @@ await this.GetDurabilityProvider(context).CreateTaskOrchestrationAsync(
             catch (InvalidOperationException ex) when (ex.Message.EndsWith("already exists.")) // for older versions of DTF.AS and DTFx.Netherite
             {
                 throw new RpcException(new Status(StatusCode.AlreadyExists, $"An Orchestration instance with the ID {request.InstanceId} already exists."));
+            }
+            catch (OperationCanceledException)
+            {
+                throw new RpcException(new Status(
+                    StatusCode.Cancelled,
+                    context.CancellationToken.IsCancellationRequested
+                    ? $"Create instance request cancelled for instance ID {request.InstanceId}"
+                    : $"Create instance request exceeded timeout of {this.extension.Options.OrchestrationCreationRequestTimeoutInSeconds} seconds " +
+                      $"for instance ID {request.InstanceId} while waiting for the termination of the existing instance with this instance ID."));
             }
             catch (Exception ex)
             {
@@ -112,12 +144,6 @@ await this.GetDurabilityProvider(context).CreateTaskOrchestrationAsync(
             }
         }
 
-        private OrchestrationStatus[] GetStatusesNotToOverride()
-        {
-            OverridableStates overridableStates = this.extension.Options.OverridableExistingInstanceStates;
-            return overridableStates.ToDedupeStatuses();
-        }
-
         public async override Task<P.RaiseEventResponse> RaiseEvent(P.RaiseEventRequest request, ServerCallContext context)
         {
             bool throwStatusExceptionsOnRaiseEvent = this.extension.Options.ThrowStatusExceptionsOnRaiseEvent ?? this.extension.DefaultDurabilityProvider.CheckStatusBeforeRaiseEvent;