Partial Purge Support (#400)

* Enhance PurgeResult to include completion status and update related gRPC response handling. Added new AbandonTask RPCs in orchestrator_service.proto. Updated versions.txt with latest proto file source.

* Update ShimDurableTaskClientTests to assert IsComplete property is not false in PurgeResult

* Add end-to-end tests for PurgeInstance and PurgeInstances_WithFilter methods in GrpcDurableTaskClientIntegrationTests. These tests verify the correct behavior of instance purging and ensure instances are removed as expected.

Author: YunchuWang

CHANGELOG.md

@@ -5,7 +5,8 @@
 - Introduce default version setting to DurableTaskClient and expose to orchestrator ([#393](https://github.com/microsoft/durabletask-dotnet/pull/393))
 - Add support for local credential types in DTS libraries ([#396](https://github.com/microsoft/durabletask-dotnet/pull/396))
 - Add utility for easier version comparison in orchestration context ([#394](https://github.com/microsoft/durabletask-dotnet/pull/394))
-- Add tags support for orchestrations ([#397])(https://github.com/microsoft/durabletask-dotnet/pull/397)
+- Add tags support for orchestrations ([#397](https://github.com/microsoft/durabletask-dotnet/pull/397))
+- Add support for versioning in the gRPC worker ([#401](https://github.com/microsoft/durabletask-dotnet/pull/401))
 
 ## v1.8.1
 

src/Abstractions/TaskOrchestrationContext.cs

@@ -1,6 +1,7 @@
 // Copyright (c) Microsoft Corporation.
 // Licensed under the MIT License.
 
+using Microsoft.DurableTask.Abstractions;
 using Microsoft.DurableTask.Entities;
 using Microsoft.Extensions.Logging;
 
@@ -421,32 +422,7 @@ public virtual ILogger CreateReplaySafeLogger<T>()
     /// <returns>True if the orchestration's version is greater than the provided version, false otherwise.</returns>
     public virtual int CompareVersionTo(string version)
     {
-        // Both versions are empty, treat as equal.
-        if (string.IsNullOrWhiteSpace(this.Version) && string.IsNullOrWhiteSpace(version))
-        {
-            return 0;
-        }
-
-        // An empty version in the context is always less than a defined version in the parameter.
-        if (string.IsNullOrWhiteSpace(this.Version))
-        {
-            return -1;
-        }
-
-        // An empty version in the parameter is always less than a defined version in the context.
-        if (string.IsNullOrWhiteSpace(version))
-        {
-            return 1;
-        }
-
-        // If both versions use the .NET Version class, return that comparison.
-        if (System.Version.TryParse(this.Version, out Version contextVersion) && System.Version.TryParse(version, out Version otherVersion))
-        {
-            return contextVersion.CompareTo(otherVersion);
-        }
-
-        // If we have gotten to here, we don't know the syntax of the versions we are comparing, use a string comparison as a final check.
-        return string.Compare(this.Version, version, StringComparison.OrdinalIgnoreCase);
+        return TaskOrchestrationVersioningUtils.CompareVersions(this.Version, version);
     }
 
     class ReplaySafeLogger : ILogger

src/Abstractions/TaskOrchestrationVersioningUtils.cs

@@ -0,0 +1,54 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+namespace Microsoft.DurableTask.Abstractions;
+
+/// <summary>
+/// Utilities for handling Orchestration/Task versioning operations.
+/// </summary>
+public static class TaskOrchestrationVersioningUtils
+{
+    /// <summary>
+    /// Compare two versions to each other.
+    /// </summary>
+    /// <remarks>
+    /// This method's comparison is handled in the following order:
+    ///   1. The versions are checked if they are empty (non-versioned). Both being empty signifies equality.
+    ///   2. If sourceVersion is empty but otherVersion is defined, this is treated as the source being less than the other.
+    ///   3. If otherVersion is empty but sourceVersion is defined, this is treated as the source being greater than the other.
+    ///   4. Both versions are attempted to be parsed into System.Version and compared as such.
+    ///   5. If all else fails, a direct string comparison is done between the versions.
+    /// </remarks>
+    /// <param name="sourceVersion">The source version that will be compared against the other version.</param>
+    /// <param name="otherVersion">The other version to compare against.</param>
+    /// <returns>An int representing how sourceVersion compares to otherVersion.</returns>
+    public static int CompareVersions(string sourceVersion, string otherVersion)
+    {
+        // Both versions are empty, treat as equal.
+        if (string.IsNullOrWhiteSpace(sourceVersion) && string.IsNullOrWhiteSpace(otherVersion))
+        {
+            return 0;
+        }
+
+        // An empty version in the context is always less than a defined version in the parameter.
+        if (string.IsNullOrWhiteSpace(sourceVersion))
+        {
+            return -1;
+        }
+
+        // An empty version in the parameter is always less than a defined version in the context.
+        if (string.IsNullOrWhiteSpace(otherVersion))
+        {
+            return 1;
+        }
+
+        // If both versions use the .NET Version class, return that comparison.
+        if (System.Version.TryParse(sourceVersion, out Version parsedSourceVersion) && System.Version.TryParse(otherVersion, out Version parsedOtherVersion))
+        {
+            return parsedSourceVersion.CompareTo(parsedOtherVersion);
+        }
+
+        // If we have gotten to here, we don't know the syntax of the versions we are comparing, use a string comparison as a final check.
+        return string.Compare(sourceVersion, otherVersion, StringComparison.OrdinalIgnoreCase);
+    }
+}

src/Client/Core/DurableTaskClient.cs

@@ -399,6 +399,19 @@ public virtual Task<PurgeResult> PurgeAllInstancesAsync(
         throw new NotSupportedException($"{this.GetType()} does not support purging of orchestration instances.");
     }
 
+    /// <summary>
+    /// Abandons a task orchestration work item to be re-enqueued at a later time.
+    /// </summary>
+    /// <param name="completionToken">The completion token that was given when the WorkItem was fetched.</param>
+    /// <param name="cancellation">
+    /// A <see cref="CancellationToken"/> that can be used to cancel the purge operation.
+    /// </param>
+    /// <returns>A task that completes when the WorkItem has been abandoned.</returns>
+    public virtual Task AbandonOrchestrationTask(string completionToken, CancellationToken cancellation = default)
+    {
+        throw new NotSupportedException($"{this.GetType()} does not support abandoning task orchestration work items.");
+    }
+
     // TODO: Create task hub
 
     // TODO: Delete task hub

src/Client/Core/PurgeResult.cs

@@ -18,9 +18,29 @@ public PurgeResult(int count)
         this.PurgedInstanceCount = count;
     }
 
+    /// <summary>
+    /// Initializes a new instance of the <see cref="PurgeResult" /> class.
+    /// </summary>
+    /// <param name="count">The number of instances deleted.</param>
+    /// <param name="isComplete">A value indicating whether the purge operation is complete.
+    /// If true, the purge operation is complete. All instances were purged.
+    /// If false, not all instances were purged. Please purge again.
+    /// If null, whether or not all instances were purged is undefined.</param>
+    public PurgeResult(int count, bool? isComplete)
+        : this(count)
+    {
+        this.IsComplete = isComplete;
+    }
+
     /// <summary>
     /// Gets the number of purged instances.
     /// </summary>
     /// <value>The number of purged instances.</value>
     public int PurgedInstanceCount { get; }
+
+    /// <summary>
+    /// Gets a value indicating whether the purge operation is complete.
+    /// </summary>
+    /// <value>A value indicating whether the purge operation is complete.</value>
+    public bool? IsComplete { get; }
 }

src/Client/Grpc/GrpcDurableTaskClient.cs

@@ -396,6 +396,17 @@ public override Task<PurgeResult> PurgeAllInstancesAsync(
         return this.PurgeInstancesCoreAsync(request, cancellation);
     }
 
+    /// <inheritdoc/>
+    public override async Task AbandonOrchestrationTask(string completionToken, CancellationToken cancellation = default)
+    {
+        P.AbandonOrchestrationTaskRequest request = new()
+        {
+            CompletionToken = completionToken,
+        };
+
+        await this.sidecarClient.AbandonTaskOrchestratorWorkItemAsync(request, cancellationToken: cancellation);
+    }
+
     static AsyncDisposable GetCallInvoker(GrpcDurableTaskClientOptions options, out CallInvoker callInvoker)
     {
         if (options.Channel is GrpcChannel c)
@@ -446,7 +457,7 @@ async Task<PurgeResult> PurgeInstancesCoreAsync(
         {
             P.PurgeInstancesResponse response = await this.sidecarClient.PurgeInstancesAsync(
                 request, cancellationToken: cancellation);
-            return new PurgeResult(response.DeletedInstanceCount);
+            return new PurgeResult(response.DeletedInstanceCount, response.IsComplete);
         }
         catch (RpcException e) when (e.StatusCode == StatusCode.Cancelled)
         {

src/Grpc/orchestrator_service.proto

@@ -463,6 +463,7 @@ message PurgeInstanceFilter {
 
 message PurgeInstancesResponse {
     int32 deletedInstanceCount = 1;
+    google.protobuf.BoolValue isComplete = 2;
 }
 
 message CreateTaskHubRequest {
@@ -617,6 +618,30 @@ message StartNewOrchestrationAction {
     google.protobuf.Timestamp scheduledTime = 5;
 }
 
+message AbandonActivityTaskRequest {
+    string completionToken = 1;
+}
+
+message AbandonActivityTaskResponse {
+    // Empty.
+}
+
+message AbandonOrchestrationTaskRequest {
+    string completionToken = 1;
+}
+
+message AbandonOrchestrationTaskResponse {
+    // Empty.
+}
+
+message AbandonEntityTaskRequest {
+    string completionToken = 1;
+}
+
+message AbandonEntityTaskResponse {
+    // Empty.
+}
+
 service TaskHubSidecarService {
     // Sends a hello request to the sidecar service.
     rpc Hello(google.protobuf.Empty) returns (google.protobuf.Empty);
@@ -678,6 +703,15 @@ service TaskHubSidecarService {
 
     // clean entity storage
     rpc CleanEntityStorage(CleanEntityStorageRequest) returns (CleanEntityStorageResponse);
+
+    // Abandons a single work item
+    rpc AbandonTaskActivityWorkItem(AbandonActivityTaskRequest) returns (AbandonActivityTaskResponse);
+
+    // Abandon an orchestration work item
+    rpc AbandonTaskOrchestratorWorkItem(AbandonOrchestrationTaskRequest) returns (AbandonOrchestrationTaskResponse);
+
+    // Abandon an entity work item
+    rpc AbandonTaskEntityWorkItem(AbandonEntityTaskRequest) returns (AbandonEntityTaskResponse);
 }
 
 message GetWorkItemsRequest {

src/Grpc/versions.txt

@@ -1,2 +1,2 @@
-# The following files were downloaded from branch main at 2025-03-19 19:55:31 UTC
-https://raw.githubusercontent.com/microsoft/durabletask-protobuf/4792f47019ab2b3e9ea979fb4af72427a4144c51/protos/orchestrator_service.proto
+# The following files were downloaded from branch main at 2025-03-24 23:37:31 UTC
+https://raw.githubusercontent.com/microsoft/durabletask-protobuf/c85ef11430ff8e10e21105abb545b0803bb86c66/protos/orchestrator_service.proto

src/Worker/Core/DependencyInjection/DurableTaskWorkerBuilderExtensions.cs

@@ -4,6 +4,7 @@
 using Microsoft.DurableTask.Worker.Hosting;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Options;
+using static Microsoft.DurableTask.Worker.DurableTaskWorkerOptions;
 
 namespace Microsoft.DurableTask.Worker;
 
@@ -86,4 +87,25 @@ public static IDurableTaskWorkerBuilder UseBuildTarget<TTarget, TOptions>(this I
             });
         return builder;
     }
+
+    /// <summary>
+    /// Configures the versioning options for this builder.
+    /// </summary>
+    /// <param name="builder">The builder to set the builder target for.</param>
+    /// <param name="versionOptions">The collection of options specified for versioning the worker.</param>
+    /// <returns>The original builder, for call chaining.</returns>
+    public static IDurableTaskWorkerBuilder UseVersioning(this IDurableTaskWorkerBuilder builder, VersioningOptions versionOptions)
+    {
+        Check.NotNull(builder);
+        builder.Configure(options =>
+        {
+            options.Versioning = new VersioningOptions
+            {
+                Version = versionOptions.Version,
+                MatchStrategy = versionOptions.MatchStrategy,
+                FailureStrategy = versionOptions.FailureStrategy,
+            };
+        });
+        return builder;
+    }
 }