Introduce orchestration versioning into worker

This commit gives the worker several ways to configure versioning.
Included are, setting the version, how to match against incoming
versions, and how to react to a failed match.

Signed-off-by: halspang <[email protected]>

Author: halspang

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/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)

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;
+    }
 }

src/Worker/Core/DurableTaskWorkerOptions.cs

@@ -10,8 +10,51 @@ namespace Microsoft.DurableTask.Worker;
 /// </summary>
 public class DurableTaskWorkerOptions
 {
+    readonly VersioningOptions versioning = new();
     DataConverter dataConverter = JsonDataConverter.Default;
 
+    /// <summary>
+    /// Defines the version matching strategy for the Durable Task worker.
+    /// </summary>
+    public enum VersionMatchStrategy
+    {
+        /// <summary>
+        /// Ignore Orchestration version, all work received is processed.
+        /// </summary>
+        None = 0,
+
+        /// <summary>
+        /// Worker will only process Tasks from Orchestrations with the same version as the worker.
+        /// </summary>
+        Strict = 1,
+
+        /// <summary>
+        /// Worker will process Tasks from Orchestrations whose version is less than or equal to the worker.
+        /// </summary>
+        CurrentOrOlder = 2,
+    }
+
+    /// <summary>
+    /// Defines the versioning failure strategy for the Durable Task worker.
+    /// </summary>
+    public enum VersionFailureStrategy
+    {
+        /// <summary>
+        /// Do not change the orchestration state if the version does not adhere to the matching strategy.
+        /// </summary>
+        Reject = 0,
+
+        /// <summary>
+        /// Suspend the orchestration if the version does not adhere to the matching strategy.
+        /// </summary>
+        Suspend = 1,
+
+        /// <summary>
+        /// Fail the orchestration if the version does not adhere to the matching strategy.
+        /// </summary>
+        Fail = 2,
+    }
+
     /// <summary>
     /// Gets or sets the data converter. Default value is <see cref="JsonDataConverter.Default" />.
     /// </summary>
@@ -93,6 +136,38 @@ public DataConverter DataConverter
     /// </remarks>
     public ConcurrencyOptions Concurrency { get; } = new();
 
+    /// <summary>
+    /// Gets or sets the versioning options for the Durable Task worker.
+    /// </summary>
+    /// <remarks>
+    /// Worker versioning controls how a worker will handle orchestrations of different versions. Defining both the
+    /// version of the worker, the versions that can be worked on, and what to do in case a version does not comply
+    /// with the given options.
+    /// </remarks>
+    public VersioningOptions Versioning
+    {
+        get => this.versioning;
+        set
+        {
+            if (value is not null)
+            {
+                this.IsVersioningSet = true;
+                this.Versioning.Version = value.Version;
+                this.Versioning.MatchStrategy = value.MatchStrategy;
+                this.Versioning.FailureStrategy = value.FailureStrategy;
+            }
+            else
+            {
+                this.IsVersioningSet = false;
+            }
+        }
+    }
+
+    /// <summary>
+    /// Gets a value indicating whether versioning is explicitly set or not.
+    /// </summary>
+    public bool IsVersioningSet { get; internal set; }
+
     /// <summary>
     /// Gets a value indicating whether <see cref="DataConverter" /> was explicitly set or not.
     /// </summary>
@@ -116,6 +191,7 @@ internal void ApplyTo(DurableTaskWorkerOptions other)
             other.DataConverter = this.DataConverter;
             other.MaximumTimerInterval = this.MaximumTimerInterval;
             other.EnableEntitySupport = this.EnableEntitySupport;
+            other.Versioning = this.Versioning;
         }
     }
 
@@ -139,4 +215,28 @@ public class ConcurrencyOptions
         /// </summary>
         public int MaximumConcurrentEntityWorkItems { get; set; } = 100 * Environment.ProcessorCount;
     }
+
+    /// <summary>
+    /// Options for the Durable Task worker versioning.
+    /// </summary>
+    public class VersioningOptions
+    {
+        /// <summary>
+        /// Gets or sets the version of orchestrations that the worker can work on.
+        /// </summary>
+        public string Version { get; set; } = string.Empty;
+
+        /// <summary>
+        /// Gets or sets the versioning strategy for the Durable Task worker.
+        /// </summary>
+        public VersionMatchStrategy MatchStrategy { get; set; } = VersionMatchStrategy.None;
+
+        /// <summary>
+        /// Gets or sets the versioning failure strategy for the Durable Task worker.
+        /// </summary>
+        /// <remarks>
+        /// If the version matching strategy is set to <see cref="VersionMatchStrategy.None"/>, this value has no effect.
+        /// </remarks>
+        public VersionFailureStrategy FailureStrategy { get; set; } = VersionFailureStrategy.Reject;
+    }
 }