Resolve merge conflict with main branch

- Rebased on latest main (f3a7fca)
- Added dedupe status tests from main alongside suspend/resume tests
- All 15 tests now present (5 dedupe + 5 suspend/resume + 5 original)
- Resolved import conflicts (added Grpc.Core using statements)
- All builds passing

Co-authored-by: YunchuWang <[email protected]>

Author: Copilot

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/Analyzers/AnalyzerReleases.Shipped.md

@@ -1,6 +1,15 @@
 ; Shipped analyzer releases
 ; https://github.com/dotnet/roslyn/blob/main/src/RoslynAnalyzers/Microsoft.CodeAnalysis.Analyzers/ReleaseTrackingAnalyzers.Help.md
 
+## Release 0.2.0
+
+### New Rules
+
+Rule ID | Category | Severity | Notes
+--------|----------|----------|-------
+DURABLE2003 | Activity | Warning | **FunctionNotFoundAnalyzer**: Warns when an activity function call references a name that does not match any defined activity in the compilation.
+DURABLE2004 | Orchestration | Warning | **FunctionNotFoundAnalyzer**: Warns when a sub-orchestration call references a name that does not match any defined orchestrator in the compilation.
+
 ## Release 0.1.0
 
 ### New Rules

src/Analyzers/AnalyzerReleases.Unshipped.md

@@ -4,6 +4,4 @@
 ### New Rules
 
 Rule ID | Category | Severity | Notes
---------|----------|----------|-------
-DURABLE2003 | Activity | Warning | **FunctionNotFoundAnalyzer**: Warns when an activity function call references a name that does not match any defined activity in the compilation.
-DURABLE2004 | Orchestration | Warning | **FunctionNotFoundAnalyzer**: Warns when a sub-orchestration call references a name that does not match any defined orchestrator in the compilation.
+--------|----------|----------|-------

src/Analyzers/Analyzers.csproj

@@ -11,7 +11,7 @@
   </PropertyGroup>
 
   <PropertyGroup>
-    <VersionPrefix>0.1.0</VersionPrefix>
+    <VersionPrefix>0.2.0</VersionPrefix>
     <VersionSuffix></VersionSuffix>
     <PackageDescription>.NET Analyzers for the Durable Task SDK.</PackageDescription>
     <NeutralLanguage>en</NeutralLanguage>

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/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs

@@ -2,7 +2,9 @@
 // Licensed under the MIT License.
 
 using System.IO.Compression;
+using System.Net;
 using System.Text;
+using Azure;
 using Azure.Core;
 using Azure.Storage.Blobs;
 using Azure.Storage.Blobs.Models;
@@ -117,20 +119,30 @@ public override async Task<string> DownloadAsync(string token, CancellationToken
 
         BlobClient blob = this.containerClient.GetBlobClient(name);
 
-        using BlobDownloadStreamingResult result = await blob.DownloadStreamingAsync(cancellationToken: cancellationToken);
-        Stream contentStream = result.Content;
-        bool isGzip = string.Equals(
-            result.Details.ContentEncoding, ContentEncodingGzip, StringComparison.OrdinalIgnoreCase);
+        try
+        {
+            using BlobDownloadStreamingResult result = await blob.DownloadStreamingAsync(cancellationToken: cancellationToken);
+            Stream contentStream = result.Content;
+            bool isGzip = string.Equals(
+                result.Details.ContentEncoding, ContentEncodingGzip, StringComparison.OrdinalIgnoreCase);
+
+            if (isGzip)
+            {
+                using GZipStream decompressed = new(contentStream, CompressionMode.Decompress);
+                using StreamReader reader = new(decompressed, Encoding.UTF8);
+                return await ReadToEndAsync(reader, cancellationToken);
+            }
 
-        if (isGzip)
+            using StreamReader uncompressedReader = new(contentStream, Encoding.UTF8);
+            return await ReadToEndAsync(uncompressedReader, cancellationToken);
+        }
+        catch (RequestFailedException ex) when (ex.Status == (int)HttpStatusCode.NotFound)
         {
-            using GZipStream decompressed = new(contentStream, CompressionMode.Decompress);
-            using StreamReader reader = new(decompressed, Encoding.UTF8);
-            return await ReadToEndAsync(reader, cancellationToken);
+            throw new InvalidOperationException(
+                $"The blob '{name}' was not found in container '{container}'. " +
+                "The payload may have been deleted or the container was never created.",
+                ex);
         }
-
-        using StreamReader uncompressedReader = new(contentStream, Encoding.UTF8);
-        return await ReadToEndAsync(uncompressedReader, cancellationToken);
     }
 
     /// <inheritdoc/>