diff --git a/src/Abstractions/TaskActivityContext.cs b/src/Abstractions/TaskActivityContext.cs index 7a622f96..63cd57b8 100644 --- a/src/Abstractions/TaskActivityContext.cs +++ b/src/Abstractions/TaskActivityContext.cs @@ -25,4 +25,10 @@ public abstract class TaskActivityContext /// Gets the unique ID of the current orchestration instance. /// public abstract string InstanceId { get; } + + /// + /// Gets the version that the activity was scheduled with, or if the activity is + /// unversioned. + /// + public virtual string Version => string.Empty; } diff --git a/src/Worker/Core/Shims/TaskActivityShim.cs b/src/Worker/Core/Shims/TaskActivityShim.cs index ae1f3bb0..37f406e5 100644 --- a/src/Worker/Core/Shims/TaskActivityShim.cs +++ b/src/Worker/Core/Shims/TaskActivityShim.cs @@ -92,5 +92,7 @@ public TaskActivityContextWrapper(TaskContext taskContext, TaskName name) public override TaskName Name => this.name; public override string InstanceId => this.innerContext.OrchestrationInstance.InstanceId; + + public override string Version => this.innerContext.Version ?? string.Empty; } } diff --git a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs index 40c8754d..7b7c41c7 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs @@ -884,10 +884,10 @@ async Task OnRunActivityAsync(P.ActivityRequest request, string completionToken, this.Logger.ReceivedActivityRequest(request.Name, request.TaskId, instance.InstanceId, inputSize); P.TaskFailureDetails? failureDetails = null; - TaskContext innerContext = new(instance); + TaskName name = new(request.Name); + TaskContext innerContext = new(instance, request.Name, request.Version ?? string.Empty, request.TaskId); innerContext.ExceptionPropertiesProvider = this.exceptionPropertiesProvider; - TaskName name = new(request.Name); string? output = null; failureDetails = EvaluateOrchestrationVersioning(this.worker.workerOptions.Versioning, request.Version, out bool versioningFailed); diff --git a/test/Grpc.IntegrationTests/OrchestrationPatterns.cs b/test/Grpc.IntegrationTests/OrchestrationPatterns.cs index 09acb365..6ae19097 100644 --- a/test/Grpc.IntegrationTests/OrchestrationPatterns.cs +++ b/test/Grpc.IntegrationTests/OrchestrationPatterns.cs @@ -791,6 +791,34 @@ public async Task OrchestrationVersionPassedThroughContext() Assert.Equal(output, $"Orchestration version: {version}"); } + [Fact] + public async Task ActivityVersionPassedThroughContext() + { + var version = "0.1"; + await using HostTestLifetime server = await this.StartWorkerAsync(b => + { + b.AddTasks(tasks => tasks + .AddOrchestratorFunc("Versioned_Orchestration", (ctx, input) => + { + return ctx.CallActivityAsync("Versioned_Activity", input); + }) + .AddActivityFunc("Versioned_Activity", (ctx, input) => + { + return $"Activity version: {ctx.Version}"; + })); + }, 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(); + + Assert.NotNull(output); + Assert.Equal(output, $"Activity version: {version}"); + } + [Fact] public async Task OrchestrationVersioning_MatchTypeNotSpecified_NoVersionFailure() { diff --git a/test/Worker/Core.Tests/Shims/TaskActivityShimTests.cs b/test/Worker/Core.Tests/Shims/TaskActivityShimTests.cs new file mode 100644 index 00000000..8bc82980 --- /dev/null +++ b/test/Worker/Core.Tests/Shims/TaskActivityShimTests.cs @@ -0,0 +1,68 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using DurableTask.Core; +using Microsoft.DurableTask.Converters; +using Microsoft.Extensions.Logging.Abstractions; + +namespace Microsoft.DurableTask.Worker.Shims; + +public class TaskActivityShimTests +{ + [Fact] + public async Task RunAsync_ScheduledWithVersion_ExposesVersionOnContext() + { + // Arrange + CapturingActivity activity = new(); + TaskActivityShim shim = new( + NullLoggerFactory.Instance, + new JsonDataConverter(), + "TestActivity", + activity); + TaskContext coreContext = new( + new OrchestrationInstance { InstanceId = Guid.NewGuid().ToString() }, + "TestActivity", + "v2", + taskId: 1); + + // Act + await shim.RunAsync(coreContext, "null"); + + // Assert + activity.CapturedVersion.Should().Be("v2"); + } + + [Fact] + public async Task RunAsync_ScheduledWithoutVersion_ExposesEmptyVersionOnContext() + { + // Arrange + CapturingActivity activity = new(); + TaskActivityShim shim = new( + NullLoggerFactory.Instance, + new JsonDataConverter(), + "TestActivity", + activity); + TaskContext coreContext = new( + new OrchestrationInstance { InstanceId = Guid.NewGuid().ToString() }, + "TestActivity", + version: null, + taskId: 1); + + // Act + await shim.RunAsync(coreContext, "null"); + + // Assert + activity.CapturedVersion.Should().Be(string.Empty); + } + + sealed class CapturingActivity : TaskActivity + { + public string? CapturedVersion { get; private set; } + + public override Task RunAsync(TaskActivityContext context, object? input) + { + this.CapturedVersion = context.Version; + return Task.FromResult(null); + } + } +}