Introduce versioning to the DurableTaskClient

This change adds a DefaultVersion to the DurableTaskClient builder
options. When no other version is set, which is currently always the
case, this value is used as the version of any new orchestration
started. The value is then passed down to workers via the
TaskOrchestrationContext.

This allows users to specify a version during their app setup and
then key off of the version during orchestration. Using conditional
logic, changes can then be made to the orchestration without harming
in-progress orchestrations.

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

Author: halspang

src/Abstractions/TaskOrchestrationContext.cs

@@ -59,6 +59,11 @@ public abstract class TaskOrchestrationContext
     /// </value>
     public abstract bool IsReplaying { get; }
 
+    /// <summary>
+    /// Gets the version.
+    /// </summary>
+    public abstract string Version { get; }
+
     /// <summary>
     /// Gets the entity feature, for interacting with entities.
     /// </summary>

src/Client/Core/DependencyInjection/DurableTaskClientBuilderExtensions.cs

@@ -89,4 +89,17 @@ public static IDurableTaskClientBuilder UseBuildTarget<TTarget, TOptions>(this I
             });
         return builder;
     }
+
+    /// <summary>
+    /// Sets the default version for this builder. This version will be applied by default to all orchestrations if set.
+    /// </summary>
+    /// <param name="builder">The builder to set the version for.</param>
+    /// <param name="version">The version that will be used as the default version.</param>
+    /// <returns>The original builder, for call chaining.</returns>
+    public static IDurableTaskClientBuilder UseDefaultVersion(this IDurableTaskClientBuilder builder, string version)
+    {
+        builder.Services.AddOptions<DurableTaskClientOptions>(builder.Name)
+            .Configure(options => options.DefaultVersion = version);
+        return builder;
+    }
 }

src/Client/Core/DurableTaskClientOptions.cs

@@ -13,6 +13,14 @@ public class DurableTaskClientOptions
     DataConverter dataConverter = JsonDataConverter.Default;
     bool enableEntitySupport;
 
+    /// <summary>
+    /// Gets or sets the version of orchestrations that will be created.
+    /// <remarks>
+    /// Currently, this is sourced from the AzureManaged client options.
+    /// </remarks>
+    /// </summary>
+    public string DefaultVersion { get; set; } = string.Empty;
+
     /// <summary>
     /// Gets or sets the data converter. Default value is <see cref="JsonDataConverter.Default" />.
     /// </summary>
@@ -95,6 +103,11 @@ internal void ApplyTo(DurableTaskClientOptions other)
             {
                 other.EnableEntitySupport = this.EnableEntitySupport;
             }
+
+            if (!string.IsNullOrWhiteSpace(this.DefaultVersion))
+            {
+                other.DefaultVersion = this.DefaultVersion;
+            }
         }
     }
 }

src/Client/Grpc/GrpcDurableTaskClient.cs

@@ -78,10 +78,20 @@ public override async Task<string> ScheduleNewOrchestrationInstanceAsync(
     {
         Check.NotEntity(this.options.EnableEntitySupport, options?.InstanceId);
 
+        var version = string.Empty;
+        if (!string.IsNullOrEmpty(orchestratorName.Version))
+        {
+            version = orchestratorName.Version;
+        }
+        else if (!string.IsNullOrEmpty(this.options.DefaultVersion))
+        {
+            version = this.options.DefaultVersion;
+        }
+
         var request = new P.CreateInstanceRequest
         {
             Name = orchestratorName.Name,
-            Version = orchestratorName.Version,
+            Version = version,
             InstanceId = options?.InstanceId ?? Guid.NewGuid().ToString("N"),
             Input = this.DataConverter.Serialize(input),
         };

test/Client/Grpc.Tests/DependencyInjection/DurableTaskClientBuilderExtensionsTests.cs

@@ -67,6 +67,20 @@ public void UseGrpc_Callback_Sets()
         options.Address.Should().BeNull();
     }
 
+    [Fact]
+    public void UseDefaultVersion_DefaultVersion_Sets()
+    {
+        ServiceCollection services = new();
+        DefaultDurableTaskClientBuilder builder = new(null, services);
+        builder.UseDefaultVersion("0.1")
+            .UseGrpc();
+
+        var provider = services.BuildServiceProvider();
+        var options = provider.GetOptions<GrpcDurableTaskClientOptions>();
+
+        options.DefaultVersion.Should().Be("0.1");
+    }
+
 #if NET6_0_OR_GREATER
     static GrpcChannel GetChannel() => GrpcChannel.ForAddress("http://localhost:9001");
 #endif

test/Grpc.IntegrationTests/IntegrationTestBase.cs

@@ -2,13 +2,13 @@
 // Licensed under the MIT License.
 
 using System.Diagnostics;
-using Microsoft.DurableTask.Tests.Logging;
 using Microsoft.DurableTask.Client;
+using Microsoft.DurableTask.Tests.Logging;
 using Microsoft.DurableTask.Worker;
+using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Logging;
 using Xunit.Abstractions;
-using Microsoft.Extensions.DependencyInjection;
 
 namespace Microsoft.DurableTask.Grpc.Tests;
 
@@ -43,9 +43,9 @@ void IDisposable.Dispose()
         GC.SuppressFinalize(this);
     }
 
-    protected async Task<HostTestLifetime> StartWorkerAsync(Action<IDurableTaskWorkerBuilder> configure)
+    protected async Task<HostTestLifetime> StartWorkerAsync(Action<IDurableTaskWorkerBuilder> configure, Action<IDurableTaskClientBuilder>? clientConfigure = null)
     {
-        IHost host = this.CreateHostBuilder(configure).Build();
+        IHost host = this.CreateHostBuilder(configure, clientConfigure).Build();
         await host.StartAsync(this.TimeoutToken);
         return new HostTestLifetime(host, this.TimeoutToken);
     }
@@ -54,7 +54,7 @@ protected async Task<HostTestLifetime> StartWorkerAsync(Action<IDurableTaskWorke
     /// Creates a <see cref="IHostBuilder"/> configured to output logs to xunit logging infrastructure.
     /// </summary>
     /// <param name="configure">Configures the durable task builder.</param>
-    protected IHostBuilder CreateHostBuilder(Action<IDurableTaskWorkerBuilder> configure)
+    protected IHostBuilder CreateHostBuilder(Action<IDurableTaskWorkerBuilder> configure, Action<IDurableTaskClientBuilder>? clientConfigure)
     {
         return Host.CreateDefaultBuilder()
             .ConfigureLogging(b =>
@@ -75,6 +75,10 @@ protected IHostBuilder CreateHostBuilder(Action<IDurableTaskWorkerBuilder> confi
                 {
                     b.UseGrpc(this.sidecarFixture.Channel);
                     b.RegisterDirectly();
+                    if (clientConfigure != null)
+                    {
+                        clientConfigure!(b);
+                    }
                 });
             });
     }

test/Grpc.IntegrationTests/OrchestrationPatterns.cs

@@ -3,11 +3,11 @@
 
 using System.Text.Json;
 using System.Text.Json.Nodes;
-using Microsoft.DurableTask.Worker;
+using Microsoft.DurableTask.Client;
 using Microsoft.DurableTask.Tests.Logging;
+using Microsoft.DurableTask.Worker;
 using Microsoft.Extensions.DependencyInjection;
 using Xunit.Abstractions;
-using Microsoft.DurableTask.Client;
 
 namespace Microsoft.DurableTask.Grpc.Tests;
 
@@ -561,8 +561,37 @@ public async Task SpecialSerialization()
         Assert.Equal("new value", output?["newProperty"]?.ToString());
     }
 
+    // TODO: Additional versioning tests
+    [Fact]
+    public async Task OrchestrationVersionPassedThroughContext()
+    {
+        var version = "0.1";
+        await using HostTestLifetime server = await this.StartWorkerAsync(b =>
+        {
+            b.AddTasks(tasks => tasks
+                .AddOrchestratorFunc<string, string>("Versioned_Orchestration", (ctx, input) =>
+                {
+                    return ctx.CallActivityAsync<string>("Versioned_Activity", ctx.Version);
+                })
+                .AddActivityFunc<string, string>("Versioned_Activity", (ctx, input) =>
+                {
+                    return $"Orchestration version: {input}";
+                }));
+        }, c =>
+        {
+            c.UseDefaultVersion(version);
+        });
+
+        var instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync("Versioned_Orchestration", input: string.Empty);
+        var result = await server.Client.WaitForInstanceCompletionAsync(instanceId, getInputsAndOutputs: true, this.TimeoutToken);
+        var output = result.ReadOutputAs<string>();
+
+        Assert.NotNull(output);
+        Assert.Equal(output, $"Orchestration version: {version}");
+
+    }
+
     // TODO: Test for multiple external events with the same name
     // TODO: Test for ContinueAsNew with external events that carry over
     // TODO: Test for catching activity exceptions of specific types
-    // TODO: Versioning tests
 }