Changing the default dedupe statuses behavior (#622)

* first commit?

* PR comments

* updating documentation

* added implementation to shim client too

* added tests for the shim client

* PR comments

* reverted the logic to not include a reuse policy in the case that dedupe statuses is null (never set by the client. this is different than explicitly making it an empty list)

* updating the tests accordingly

* PR comments

* addressing PR comments

* updated tests to check for new exception type

* slight comment update

* Adding an ArgumentException for invalid dedupe statuses (any running + terminated)

* added support to terminate existing running instances for restart

* addressing copilot comments

* addressing the PR comments and build warnings

* fixed the build errors

* missed a change in comment type in GrpcDurableTaskClient

* missed updating a comment in the tests

* returned the catching of the RpcException with status code cancelled, since even if the server no longer throws an exception like this gRPC can still throw this under the hood

---------

Co-authored-by: Sophia Tevosyan <stevosyan@microsoft.com>

Author: sophiatev

src/Client/Core/DurableTaskClient.cs

@@ -2,6 +2,7 @@
 // Licensed under the MIT License.
 
 using System.ComponentModel;
+using DurableTask.Core.Exceptions;
 using DurableTask.Core.History;
 using Microsoft.DurableTask.Client.Entities;
 using Microsoft.DurableTask.Internal;
@@ -73,10 +74,14 @@ public virtual Task<string> ScheduleNewOrchestrationInstanceAsync(
     /// <remarks>
     /// <para>All orchestrations must have a unique instance ID. You can provide an instance ID using the
     /// <paramref name="options"/> parameter or you can omit this and a random instance ID will be
-    /// generated for you automatically. If an orchestration with the specified instance ID already exists and is in a
-    /// non-terminal state (Pending, Running, etc.), then this operation may fail silently. However, if an orchestration
-    /// instance with this ID already exists in a terminal state (Completed, Terminated, Failed, etc.) then the instance
-    /// may be recreated automatically, depending on the configuration of the backend instance store.
+    /// generated for you automatically. If an orchestration with the specified instance ID already exists and its status
+    /// is not in the <see cref="StartOrchestrationOptions.DedupeStatuses"/> field of <paramref name="options"/>, then
+    /// a new orchestration may be recreated automatically, depending on the configuration of the backend instance store.
+    /// If the existing orchestration is in a non-terminal state (Pending, Running, etc.), then the orchestration will first
+    /// be terminated before the new orchestration is created.
+    /// A null <see cref="StartOrchestrationOptions.DedupeStatuses"/> field means the deduplication behavior will follow
+    /// whatever the default deduplication behavior of the backend instance store or implementation of this client is.
+    /// A non-null, empty field means that all statuses are reusable.
     /// </para><para>
     /// Orchestration instances started with this method will be created in the
     /// <see cref="OrchestrationRuntimeStatus.Pending"/> state and will transition to the
@@ -98,15 +103,24 @@ public virtual Task<string> ScheduleNewOrchestrationInstanceAsync(
     /// </param>
     /// <param name="options">The options to start the new orchestration with.</param>
     /// <param name="cancellation">
-    /// The cancellation token. This only cancels enqueueing the new orchestration to the backend. Does not cancel the
-    /// orchestration once enqueued.
+    /// The cancellation token. This only cancels enqueueing the new orchestration to the backend, or waiting for the
+    /// termination of an existing non-terminal instance if its status is not in
+    /// <see cref="StartOrchestrationOptions.DedupeStatuses"/>. Does not cancel the orchestration once enqueued.
     /// </param>
     /// <returns>
     /// A task that completes when the orchestration instance is successfully scheduled. The value of this task is
     /// the instance ID of the scheduled orchestration instance. If a non-null instance ID was provided via
     /// <paramref name="options" />, the same value will be returned by the completed task.
     /// </returns>
     /// <exception cref="ArgumentNullException">Thrown if <paramref name="orchestratorName"/> is empty.</exception>
+    /// <exception cref="OrchestrationAlreadyExistsException">If an orchestration with status in
+    /// <see cref="StartOrchestrationOptions.DedupeStatuses"/> with this instance ID already exists.</exception>
+    /// <exception cref="ArgumentException">
+    /// Thrown if <see cref="StartOrchestrationOptions.DedupeStatuses"/> contains '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 'Terminated'.
+    /// </exception>
     public abstract Task<string> ScheduleNewOrchestrationInstanceAsync(
         TaskName orchestratorName,
         object? input = null,
@@ -412,6 +426,9 @@ public virtual Task<PurgeResult> PurgeAllInstancesAsync(
     /// <para>
     /// This method restarts an existing orchestration instance. If <paramref name="restartWithNewInstanceId"/> is <c>true</c>,
     /// a new instance ID will be generated for the restarted orchestration. If <c>false</c>, the original instance ID will be reused.
+    /// If the existing instance is not in a terminal state, depending on the specific implementation, either:
+    /// 1. The existing instance will be terminated before being restarted, or
+    /// 2. An <see cref="InvalidOperationException"/> will be thrown.
     /// </para><para>
     /// The restarted orchestration will use the same input data as the original instance. If the original orchestration
     /// instance is not found, an <see cref="ArgumentException"/> will be thrown.
@@ -426,8 +443,9 @@ public virtual Task<PurgeResult> PurgeAllInstancesAsync(
     /// If <c>false</c>, the original instance ID will be reused.
     /// </param>
     /// <param name="cancellation">
-    /// The cancellation token. This only cancels enqueueing the restart request to the backend.
-    /// Does not abort restarting the orchestration once enqueued.
+    /// The cancellation token. This only cancels enqueueing the restart request to the backend,
+    /// or waiting for the termination of an existing non-terminal instance if the implementation supports
+    /// this behavior. Does not abort restarting the orchestration once enqueued.
     /// </param>
     /// <returns>
     /// A task that completes when the orchestration instance is successfully restarted.
@@ -437,7 +455,8 @@ public virtual Task<PurgeResult> PurgeAllInstancesAsync(
     /// Thrown if an orchestration with the specified <paramref name="instanceId"/> was not found. </exception>
     /// <exception cref="InvalidOperationException">
     /// Thrown when attempting to restart an instance using the same instance Id
-    /// while the instance has not yet reached a completed or terminal state. </exception>
+    /// while the instance has not yet reached a completed or terminal state, in the case that
+    /// the implementation does not support terminating existing non-terminal instances before restarting.</exception>
     /// <exception cref="NotSupportedException">
     /// Thrown if the backend does not support restart operations. </exception>
     public virtual Task<string> RestartAsync(
@@ -529,7 +548,7 @@ public virtual Task<Page<string>> ListInstanceIdsAsync(
         throw new NotSupportedException(
             $"{this.GetType()} does not support listing orchestration instance IDs filtered by completed time.");
     }
-    
+
     // TODO: Create task hub
 
     // TODO: Delete task hub
@@ -539,4 +558,4 @@ public virtual Task<Page<string>> ListInstanceIdsAsync(
     /// </summary>
     /// <returns>A <see cref="ValueTask"/> that completes when the disposal completes.</returns>
     public abstract ValueTask DisposeAsync();
-}
+}

src/Client/Core/StartOrchestrationOptionsExtensions.cs

@@ -9,14 +9,26 @@ namespace Microsoft.DurableTask.Client;
 /// Extension methods for <see cref="StartOrchestrationOptions"/> to provide type-safe deduplication status configuration.
 /// </summary>
 public static class StartOrchestrationOptionsExtensions
-{
-    public static readonly OrchestrationRuntimeStatus[] ValidDedupeStatuses = new[]
-    {
+{
+#pragma warning disable CS0618 // Type or member is obsolete - Cancelled is intentionally included for compatibility with the
+                               // Durable Task Framework
+
+    /// <summary>
+    /// The list of orchestration statuses that can be deduplicated upon a creation request.
+    /// If one of these statuses is included in the request via the <see cref="StartOrchestrationOptions.DedupeStatuses"/>
+    /// field, and an orchestration with this status and same instance ID is found, the request will fail.
+    /// </summary>
+    public static readonly IReadOnlyList<OrchestrationRuntimeStatus> ValidDedupeStatuses =
+    [
         OrchestrationRuntimeStatus.Completed,
         OrchestrationRuntimeStatus.Failed,
-        OrchestrationRuntimeStatus.Terminated,
+        OrchestrationRuntimeStatus.Terminated,
         OrchestrationRuntimeStatus.Canceled,
-    };
+        OrchestrationRuntimeStatus.Pending,
+        OrchestrationRuntimeStatus.Running,
+        OrchestrationRuntimeStatus.Suspended,
+    ];
+#pragma warning restore CS0618 // Type or member is obsolete
 
     /// <summary>
     /// Creates a new <see cref="StartOrchestrationOptions"/> with the specified deduplication statuses.

src/Client/Grpc/GrpcDurableTaskClient.cs

@@ -4,6 +4,7 @@
 using System.Collections.Immutable;
 using System.Diagnostics;
 using System.Text;
+using DurableTask.Core.Exceptions;
 using DurableTask.Core.History;
 using Google.Protobuf.WellKnownTypes;
 using Microsoft.DurableTask.Client.Entities;
@@ -73,6 +74,7 @@ public override ValueTask DisposeAsync()
     }
 
     /// <inheritdoc/>
+    // The behavior of this method when the dedupe statuses field is null depends on the server-side implementation.
     public override async Task<string> ScheduleNewOrchestrationInstanceAsync(
         TaskName orchestratorName,
         object? input = null,
@@ -124,9 +126,7 @@ public override async Task<string> ScheduleNewOrchestrationInstanceAsync(
         }
 
         // Set orchestration ID reuse policy for deduplication support
-        // Note: This requires the protobuf to support OrchestrationIdReusePolicy field
-        // If the protobuf doesn't support it yet, this will need to be updated when the protobuf is updated
-        if (options?.DedupeStatuses != null && options.DedupeStatuses.Count > 0)
+        if (options?.DedupeStatuses != null)
         {
             // Parse and validate all status strings to enum first
             ImmutableHashSet<OrchestrationRuntimeStatus> dedupeStatuses = options.DedupeStatuses
@@ -143,19 +143,30 @@ public override async Task<string> ScheduleNewOrchestrationInstanceAsync(
 
             // Convert dedupe statuses to protobuf statuses and create reuse policy
             IEnumerable<P.OrchestrationStatus> dedupeStatusesProto = dedupeStatuses.Select(s => s.ToGrpcStatus());
-            P.OrchestrationIdReusePolicy? policy = ProtoUtils.ConvertDedupeStatusesToReusePolicy(dedupeStatusesProto);
-
-            if (policy != null)
-            {
-                request.OrchestrationIdReusePolicy = policy;
-            }
+            request.OrchestrationIdReusePolicy = ProtoUtils.ConvertDedupeStatusesToReusePolicy(dedupeStatusesProto);
         }
 
         using Activity? newActivity = TraceHelper.StartActivityForNewOrchestration(request);
 
-        P.CreateInstanceResponse? result = await this.sidecarClient.StartInstanceAsync(
+        try
+        {
+            P.CreateInstanceResponse? result = await this.sidecarClient.StartInstanceAsync(
             request, cancellationToken: cancellation);
-        return result.InstanceId;
+            return result.InstanceId;
+        }
+        catch (RpcException e) when (e.StatusCode == StatusCode.AlreadyExists)
+        {
+            throw new OrchestrationAlreadyExistsException(e.Status.Detail);
+        }
+        catch (RpcException e) when (e.StatusCode == StatusCode.InvalidArgument)
+        {
+            throw new ArgumentException(e.Status.Detail);
+        }
+        catch (RpcException e) when (e.StatusCode == StatusCode.Cancelled)
+        {
+            throw new OperationCanceledException(
+                $"The {nameof(this.ScheduleNewOrchestrationInstanceAsync)} operation was canceled.", e, cancellation);
+        }
     }
 
     /// <inheritdoc/>
@@ -479,6 +490,9 @@ public override Task<PurgeResult> PurgeAllInstancesAsync(
     }
 
     /// <inheritdoc/>
+    // Whether or not this method throws a <see cref="InvalidOperationException"/> or terminates the existing instance
+    // when <paramref name="restartWithNewInstanceId"/> is <c>false</c> and the existing instance is not in a terminal state
+    // depends on the server-side implementation.
     public override async Task<string> RestartAsync(
         string instanceId,
         bool restartWithNewInstanceId = false,

src/Client/Grpc/ProtoUtils.cs

@@ -13,18 +13,21 @@ namespace Microsoft.DurableTask.Client.Grpc;
 public static class ProtoUtils
 {
     /// <summary>
-    /// Gets the terminal orchestration statuses that are commonly used for deduplication.
+    /// Gets an array of all orchestration statuses.
     /// These are the statuses that can be used in OrchestrationIdReusePolicy.
     /// </summary>
-    /// <returns>An immutable array of terminal orchestration statuses.</returns>
-    public static ImmutableArray<P.OrchestrationStatus> GetTerminalStatuses()
+    /// <returns>An immutable array of all orchestration statuses.</returns>
+    public static ImmutableArray<P.OrchestrationStatus> GetAllStatuses()
     {
 #pragma warning disable CS0618 // Type or member is obsolete - Canceled is intentionally included for compatibility
         return ImmutableArray.Create(
             P.OrchestrationStatus.Completed,
             P.OrchestrationStatus.Failed,
-            P.OrchestrationStatus.Terminated,
-            P.OrchestrationStatus.Canceled);
+            P.OrchestrationStatus.Terminated,
+            P.OrchestrationStatus.Canceled,
+            P.OrchestrationStatus.Pending,
+            P.OrchestrationStatus.Running,
+            P.OrchestrationStatus.Suspended);
 #pragma warning restore CS0618
     }
 
@@ -33,59 +36,60 @@ public static class ProtoUtils
     /// with replaceable statuses (statuses that CAN be replaced).
     /// </summary>
     /// <param name="dedupeStatuses">The orchestration statuses that should NOT be replaced. These are statuses for which an exception should be thrown if an orchestration already exists.</param>
-    /// <returns>An OrchestrationIdReusePolicy with replaceable statuses set, or null if all terminal statuses are dedupe statuses.</returns>
+    /// <returns>An OrchestrationIdReusePolicy with replaceable statuses set.</returns>
     /// <remarks>
     /// The policy uses "replaceableStatus" - these are statuses that CAN be replaced.
     /// dedupeStatuses are statuses that should NOT be replaced.
-    /// So replaceableStatus = all terminal statuses MINUS dedupeStatuses.
+    /// So replaceableStatus = all statuses MINUS dedupeStatuses.
     /// </remarks>
-    public static P.OrchestrationIdReusePolicy? ConvertDedupeStatusesToReusePolicy(
-        IEnumerable<P.OrchestrationStatus>? dedupeStatuses)
-    {
-        ImmutableArray<P.OrchestrationStatus> terminalStatuses = GetTerminalStatuses();
-        ImmutableHashSet<P.OrchestrationStatus> dedupeStatusSet = dedupeStatuses?.ToImmutableHashSet() ?? ImmutableHashSet<P.OrchestrationStatus>.Empty;
+    public static P.OrchestrationIdReusePolicy ConvertDedupeStatusesToReusePolicy(
+        IEnumerable<P.OrchestrationStatus> dedupeStatuses)
+    {
+        Check.NotNull(dedupeStatuses);
+        ImmutableArray<P.OrchestrationStatus> statuses = GetAllStatuses();
+        ImmutableHashSet<P.OrchestrationStatus> dedupeStatusSet = [.. dedupeStatuses];
 
         P.OrchestrationIdReusePolicy policy = new();
 
-        // Add terminal statuses that are NOT in dedupeStatuses as replaceable
-        foreach (P.OrchestrationStatus terminalStatus in terminalStatuses.Where(status => !dedupeStatusSet.Contains(status)))
+        // Add statuses that are NOT in dedupeStatuses as replaceable
+        foreach (P.OrchestrationStatus status in statuses.Where(status => !dedupeStatusSet.Contains(status)))
         {
-            policy.ReplaceableStatus.Add(terminalStatus);
+            policy.ReplaceableStatus.Add(status);
         }
 
-        // Only return policy if we have replaceable statuses
-        return policy.ReplaceableStatus.Count > 0 ? policy : null;
+        return policy;
     }
 
     /// <summary>
     /// Converts an OrchestrationIdReusePolicy with replaceable statuses to dedupe statuses
     /// (statuses that should NOT be replaced).
     /// </summary>
-    /// <param name="policy">The OrchestrationIdReusePolicy containing replaceable statuses.</param>
-    /// <returns>An array of orchestration statuses that should NOT be replaced, or null if all terminal statuses are replaceable.</returns>
+    /// <param name="policy">The OrchestrationIdReusePolicy containing replaceable statuses. If this parameter is null,
+    /// then null is returned.</param>
+    /// <returns>An array of orchestration statuses that should NOT be replaced, which is empty if all statuses
+    /// are replaceable.</returns>
     /// <remarks>
     /// The policy uses "replaceableStatus" - these are statuses that CAN be replaced.
     /// dedupeStatuses are statuses that should NOT be replaced (should throw exception).
-    /// So dedupeStatuses = all terminal statuses MINUS replaceableStatus.
+    /// So dedupeStatuses = all statuses MINUS replaceableStatus.
     /// </remarks>
     public static P.OrchestrationStatus[]? ConvertReusePolicyToDedupeStatuses(
         P.OrchestrationIdReusePolicy? policy)
-    {
-        if (policy == null || policy.ReplaceableStatus.Count == 0)
+    {
+        if (policy == null)
         {
             return null;
-        }
+        }
 
-        ImmutableArray<P.OrchestrationStatus> terminalStatuses = GetTerminalStatuses();
+        ImmutableArray<P.OrchestrationStatus> allStatuses = GetAllStatuses();
         ImmutableHashSet<P.OrchestrationStatus> replaceableStatusSet = policy.ReplaceableStatus.ToImmutableHashSet();
 
-        // Calculate dedupe statuses = terminal statuses - replaceable statuses
-        P.OrchestrationStatus[] dedupeStatuses = terminalStatuses
-            .Where(terminalStatus => !replaceableStatusSet.Contains(terminalStatus))
+        // Calculate dedupe statuses = all statuses - replaceable statuses
+        P.OrchestrationStatus[] dedupeStatuses = allStatuses
+            .Where(status => !replaceableStatusSet.Contains(status))
             .ToArray();
 
-        // Only return if there are dedupe statuses
-        return dedupeStatuses.Length > 0 ? dedupeStatuses : null;
+        return dedupeStatuses;
     }
 
 #pragma warning disable 0618 // Referencing Obsolete member. This is intention as we are only converting it.