From 81ff9591281d8f0348330d416215db7b88eb2624 Mon Sep 17 00:00:00 2001 From: Hal Spang Date: Fri, 12 Jun 2026 16:48:49 -0700 Subject: [PATCH] Expose scheduled version on TaskActivityContext Adds a virtual Version property to TaskActivityContext (defaulting to string.Empty for backward compatibility) and overrides it in the worker's activity context wrapper to surface the version the activity was scheduled with, mirroring TaskOrchestrationContext.Version. The gRPC worker's activity dispatch path previously constructed TaskContext with only the orchestration instance, leaving the version empty even though it travels on ActivityRequest. It now builds TaskContext with the request name, version, and task id so activities can read their scheduled version. Includes a unit test for the shim wrapper and an end-to-end integration test asserting the activity observes its scheduled version. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- src/Abstractions/TaskActivityContext.cs | 6 ++ src/Worker/Core/Shims/TaskActivityShim.cs | 2 + .../Grpc/GrpcDurableTaskWorker.Processor.cs | 4 +- .../OrchestrationPatterns.cs | 28 ++++++++ .../Core.Tests/Shims/TaskActivityShimTests.cs | 68 +++++++++++++++++++ 5 files changed, 106 insertions(+), 2 deletions(-) create mode 100644 test/Worker/Core.Tests/Shims/TaskActivityShimTests.cs 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); + } + } +}