Support dedup status when starting orchestration (#542)

Author: YunchuWang

src/Abstractions/TaskOptions.cs

@@ -134,4 +134,12 @@ public record StartOrchestrationOptions(string? InstanceId = null, DateTimeOffse
     /// Gets the version to associate with the orchestration instance.
     /// </summary>
     public TaskVersion? Version { get; init; }
+
+    /// <summary>
+    /// Gets the orchestration runtime statuses that should be considered for deduplication.
+    /// </summary>
+    /// <remarks>
+    /// For type-safe usage, use the WithDedupeStatuses extension method.
+    /// </remarks>
+    public IReadOnlyList<string>? DedupeStatuses { get; init; }
 }

src/Client/Core/StartOrchestrationOptionsExtensions.cs

@@ -0,0 +1,36 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+using System.Linq;
+
+namespace Microsoft.DurableTask.Client;
+
+/// <summary>
+/// Extension methods for <see cref="StartOrchestrationOptions"/> to provide type-safe deduplication status configuration.
+/// </summary>
+public static class StartOrchestrationOptionsExtensions
+{
+    public static readonly OrchestrationRuntimeStatus[] ValidDedupeStatuses = new[]
+    {
+        OrchestrationRuntimeStatus.Completed,
+        OrchestrationRuntimeStatus.Failed,
+        OrchestrationRuntimeStatus.Terminated,
+        OrchestrationRuntimeStatus.Canceled,
+    };
+
+    /// <summary>
+    /// Creates a new <see cref="StartOrchestrationOptions"/> with the specified deduplication statuses.
+    /// </summary>
+    /// <param name="options">The base options to extend.</param>
+    /// <param name="dedupeStatuses">The orchestration runtime statuses that should be considered for deduplication.</param>
+    /// <returns>A new <see cref="StartOrchestrationOptions"/> instance with the deduplication statuses set.</returns>
+    public static StartOrchestrationOptions WithDedupeStatuses(
+        this StartOrchestrationOptions options,
+        params OrchestrationRuntimeStatus[] dedupeStatuses)
+    {
+        return options with
+        {
+            DedupeStatuses = dedupeStatuses.Select(s => s.ToString()).ToList(),
+        };
+    }
+}

src/Client/Grpc/GrpcDurableTaskClient.cs

@@ -1,6 +1,7 @@
 // Copyright (c) Microsoft Corporation.
 // Licensed under the MIT License.
 
+using System.Collections.Immutable;
 using System.Diagnostics;
 using System.Text;
 using DurableTask.Core.History;
@@ -91,7 +92,7 @@ public override async Task<string> ScheduleNewOrchestrationInstanceAsync(
             version = this.options.DefaultVersion;
         }
 
-        var request = new P.CreateInstanceRequest
+        P.CreateInstanceRequest request = new()
         {
             Name = orchestratorName.Name,
             Version = version,
@@ -122,6 +123,34 @@ public override async Task<string> ScheduleNewOrchestrationInstanceAsync(
             request.ScheduledStartTimestamp = Timestamp.FromDateTimeOffset(startAt.Value.ToUniversalTime());
         }
 
+        // 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)
+        {
+            // Parse and validate all status strings to enum first
+            ImmutableHashSet<OrchestrationRuntimeStatus> dedupeStatuses = options.DedupeStatuses
+                .Select(s =>
+                {
+                    if (!System.Enum.TryParse<OrchestrationRuntimeStatus>(s, ignoreCase: true, out OrchestrationRuntimeStatus status))
+                    {
+                        throw new ArgumentException(
+                            $"Invalid orchestration runtime status: '{s}' for deduplication.");
+                    }
+
+                    return status;
+                }).ToImmutableHashSet();
+
+            // 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;
+            }
+        }
+
         using Activity? newActivity = TraceHelper.StartActivityForNewOrchestration(request);
 
         P.CreateInstanceResponse? result = await this.sidecarClient.StartInstanceAsync(
@@ -405,7 +434,7 @@ public override async Task<string> RestartAsync(
         Check.NotNullOrEmpty(instanceId);
         Check.NotEntity(this.options.EnableEntitySupport, instanceId);
 
-        var request = new P.RestartInstanceRequest
+        P.RestartInstanceRequest request = new P.RestartInstanceRequest
         {
             InstanceId = instanceId,
             RestartWithNewInstanceId = restartWithNewInstanceId,
@@ -441,7 +470,7 @@ public override async Task RewindInstanceAsync(
         Check.NotNullOrEmpty(instanceId);
         Check.NotEntity(this.options.EnableEntitySupport, instanceId);
 
-        var request = new P.RewindInstanceRequest
+        P.RewindInstanceRequest request = new P.RewindInstanceRequest
         {
             InstanceId = instanceId,
             Reason = reason,
@@ -573,7 +602,7 @@ async Task<PurgeResult> PurgeInstancesCoreAsync(
 
     OrchestrationMetadata CreateMetadata(P.OrchestrationState state, bool includeInputsAndOutputs)
     {
-        var metadata = new OrchestrationMetadata(state.Name, state.InstanceId)
+        OrchestrationMetadata metadata = new OrchestrationMetadata(state.Name, state.InstanceId)
         {
             CreatedAt = state.CreatedTimestamp.ToDateTimeOffset(),
             LastUpdatedAt = state.LastUpdatedTimestamp.ToDateTimeOffset(),

src/Client/Grpc/ProtoUtils.cs

@@ -1,15 +1,93 @@
 // Copyright (c) Microsoft Corporation.
 // Licensed under the MIT License.
 
+using System.Collections.Immutable;
+using System.Linq;
 using P = Microsoft.DurableTask.Protobuf;
 
 namespace Microsoft.DurableTask.Client.Grpc;
 
 /// <summary>
 /// Protobuf helpers and utilities.
 /// </summary>
-static class ProtoUtils
+public static class ProtoUtils
 {
+    /// <summary>
+    /// Gets the terminal orchestration statuses that are commonly used for deduplication.
+    /// 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()
+    {
+#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);
+#pragma warning restore CS0618
+    }
+
+    /// <summary>
+    /// Converts dedupe statuses (statuses that should NOT be replaced) to an OrchestrationIdReusePolicy
+    /// 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>
+    /// <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.
+    /// </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;
+
+        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)))
+        {
+            policy.ReplaceableStatus.Add(terminalStatus);
+        }
+
+        // Only return policy if we have replaceable statuses
+        return policy.ReplaceableStatus.Count > 0 ? policy : null;
+    }
+
+    /// <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>
+    /// <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.
+    /// </remarks>
+    public static P.OrchestrationStatus[]? ConvertReusePolicyToDedupeStatuses(
+        P.OrchestrationIdReusePolicy? policy)
+    {
+        if (policy == null || policy.ReplaceableStatus.Count == 0)
+        {
+            return null;
+        }
+
+        ImmutableArray<P.OrchestrationStatus> terminalStatuses = GetTerminalStatuses();
+        ImmutableHashSet<P.OrchestrationStatus> replaceableStatusSet = policy.ReplaceableStatus.ToImmutableHashSet();
+
+        // Calculate dedupe statuses = terminal statuses - replaceable statuses
+        P.OrchestrationStatus[] dedupeStatuses = terminalStatuses
+            .Where(terminalStatus => !replaceableStatusSet.Contains(terminalStatus))
+            .ToArray();
+
+        // Only return if there are dedupe statuses
+        return dedupeStatuses.Length > 0 ? dedupeStatuses : null;
+    }
+
 #pragma warning disable 0618 // Referencing Obsolete member. This is intention as we are only converting it.
     /// <summary>
     /// Converts <see cref="OrchestrationRuntimeStatus" /> to <see cref="P.OrchestrationStatus" />.

src/Client/OrchestrationServiceClientShim/ShimDurableTaskClient.cs

@@ -7,6 +7,7 @@
 using DurableTask.Core;
 using DurableTask.Core.History;
 using DurableTask.Core.Query;
+using Microsoft.DurableTask.Client;
 using Microsoft.DurableTask.Client.Entities;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Options;
@@ -192,7 +193,23 @@ public override async Task<string> ScheduleNewOrchestrationInstanceAsync(
             },
         };
 
-        await this.Client.CreateTaskOrchestrationAsync(message);
+        Core.OrchestrationStatus[]? dedupeStatuses = null;
+        if (options?.DedupeStatuses != null && options.DedupeStatuses.Count > 0)
+        {
+            dedupeStatuses = options.DedupeStatuses
+                .Select(s =>
+                {
+                    if (!Enum.TryParse<OrchestrationRuntimeStatus>(s, ignoreCase: true, out var status))
+                    {
+                        throw new ArgumentException(
+                            $"Invalid orchestration runtime status: '{s}' for deduplication.");
+                    }
+                    return status.ConvertToCore();
+                })
+                .ToArray();
+        }
+
+        await this.Client.CreateTaskOrchestrationAsync(message, dedupeStatuses);
         return instanceId;
     }
 
@@ -303,7 +320,7 @@ public override async Task<string> RestartAsync(
             },
         };
 
-        await this.Client.CreateTaskOrchestrationAsync(message);
+        await this.Client.CreateTaskOrchestrationAsync(message, dedupeStatuses: null);
         return newInstanceId;
     }
 

src/InProcessTestHost/Sidecar/Grpc/TaskHubGrpcServer.cs

@@ -1,14 +1,17 @@
-// Copyright (c) Microsoft Corporation.
+// Copyright (c) Microsoft Corporation.
 // Licensed under the MIT License.
 
 using System.Collections.Concurrent;
 using System.Diagnostics;
 using System.Globalization;
+using System.Linq;
 using DurableTask.Core;
+using DurableTask.Core.Exceptions;
 using DurableTask.Core.History;
 using DurableTask.Core.Query;
 using Google.Protobuf.WellKnownTypes;
 using Grpc.Core;
+using Microsoft.DurableTask.Client.Grpc;
 using Microsoft.DurableTask.Testing.Sidecar.Dispatcher;
 using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Logging;
@@ -202,6 +205,18 @@ async Task WaitForWorkItemClientConnection()
 
         try
         {
+            // Convert OrchestrationIdReusePolicy to dedupeStatuses
+            // 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
+            OrchestrationStatus[]? dedupeStatuses = null;
+            P.OrchestrationStatus[]? dedupeStatusesProto = ProtoUtils.ConvertReusePolicyToDedupeStatuses(request.OrchestrationIdReusePolicy);
+            if (dedupeStatusesProto != null)
+            {
+                // Convert protobuf statuses to Core.OrchestrationStatus
+                dedupeStatuses = dedupeStatusesProto.Select(s => (OrchestrationStatus)s).ToArray();
+            }
+
             await this.client.CreateTaskOrchestrationAsync(
                 new TaskMessage
                 {
@@ -216,7 +231,14 @@ await this.client.CreateTaskOrchestrationAsync(
                             : null
                     },
                     OrchestrationInstance = instance,
-                });
+                },
+                dedupeStatuses);
+        }
+        catch (OrchestrationAlreadyExistsException e)
+        {
+            // Convert to gRPC exception
+            this.log.LogWarning(e, "Orchestration with ID {InstanceId} already exists", instance.InstanceId);
+            throw new RpcException(new Status(StatusCode.AlreadyExists, e.Message));
         }
         catch (Exception e)
         {

test/Abstractions.Tests/Abstractions.Tests.csproj

@@ -6,6 +6,7 @@
 
   <ItemGroup>
     <ProjectReference Include="$(SrcRoot)Abstractions/Abstractions.csproj" />
+    <ProjectReference Include="$(SrcRoot)Client/Core/Client.csproj" />
   </ItemGroup>
 
 </Project>