From ec8e0d571b67282db83155100e03fb3b4179b7e2 Mon Sep 17 00:00:00 2001 From: Hal Spang Date: Tue, 30 Sep 2025 11:24:52 -0700 Subject: [PATCH 1/2] Add version check to activities Activities were previously not checking the versioning status of the worker. This was because it was assumed that the orchestration level check would satisfy the activity. However, when multiple workers of the same version are present, the activities can be sent to workers of the incorrect version. This change adds the orchestration version to the activity to allow for versioning checks. Signed-off-by: Hal Spang --- .../Shims/TaskOrchestrationContextWrapper.cs | 6 +- .../Grpc/GrpcDurableTaskWorker.Processor.cs | 55 +++++++++++++------ 2 files changed, 40 insertions(+), 21 deletions(-) diff --git a/src/Worker/Core/Shims/TaskOrchestrationContextWrapper.cs b/src/Worker/Core/Shims/TaskOrchestrationContextWrapper.cs index 5bf3fbe5..c7090e0b 100644 --- a/src/Worker/Core/Shims/TaskOrchestrationContextWrapper.cs +++ b/src/Worker/Core/Shims/TaskOrchestrationContextWrapper.cs @@ -154,7 +154,7 @@ public override async Task CallActivityAsync( { return await this.innerContext.ScheduleTask( name.Name, - name.Version, + this.innerContext.Version, options: ScheduleTaskOptions.CreateBuilder() .WithRetryOptions(policy.ToDurableTaskCoreRetryOptions()) .WithTags(tags) @@ -166,7 +166,7 @@ public override async Task CallActivityAsync( return await this.InvokeWithCustomRetryHandler( () => this.innerContext.ScheduleTask( name.Name, - name.Version, + this.innerContext.Version, options: ScheduleTaskOptions.CreateBuilder() .WithTags(tags) .Build(), @@ -179,7 +179,7 @@ public override async Task CallActivityAsync( { return await this.innerContext.ScheduleTask( name.Name, - name.Version, + this.innerContext.Version, options: ScheduleTaskOptions.CreateBuilder() .WithTags(tags) .Build(), diff --git a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs index 20cd6309..3e921252 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs @@ -759,37 +759,56 @@ async Task OnRunActivityAsync(P.ActivityRequest request, string completionToken, OrchestrationInstance instance = request.OrchestrationInstance.ToCore(); string rawInput = request.Input; - int inputSize = rawInput != null ? Encoding.UTF8.GetByteCount(rawInput) : 0; this.Logger.ReceivedActivityRequest(request.Name, request.TaskId, instance.InstanceId, inputSize); + P.TaskFailureDetails? failureDetails = null; TaskContext innerContext = new(instance); TaskName name = new(request.Name); string? output = null; - P.TaskFailureDetails? failureDetails = null; - try + + failureDetails = EvaluateOrchestrationVersioning(this.worker.workerOptions.Versioning, request.Version, out bool versioningFailed); + if (!versioningFailed) { - await using AsyncServiceScope scope = this.worker.services.CreateAsyncScope(); - if (this.worker.Factory.TryCreateActivity(name, scope.ServiceProvider, out ITaskActivity? activity)) + try { - // Both the factory invocation and the RunAsync could involve user code and need to be handled as - // part of try/catch. - TaskActivity shim = this.shimFactory.CreateActivity(name, activity); - output = await shim.RunAsync(innerContext, request.Input); + await using AsyncServiceScope scope = this.worker.services.CreateAsyncScope(); + if (this.worker.Factory.TryCreateActivity(name, scope.ServiceProvider, out ITaskActivity? activity)) + { + // Both the factory invocation and the RunAsync could involve user code and need to be handled as + // part of try/catch. + TaskActivity shim = this.shimFactory.CreateActivity(name, activity); + output = await shim.RunAsync(innerContext, request.Input); + } + else + { + failureDetails = new P.TaskFailureDetails + { + ErrorType = "ActivityTaskNotFound", + ErrorMessage = $"No activity task named '{name}' was found.", + IsNonRetriable = true, + }; + } } - else + catch (Exception applicationException) { - failureDetails = new P.TaskFailureDetails - { - ErrorType = "ActivityTaskNotFound", - ErrorMessage = $"No activity task named '{name}' was found.", - IsNonRetriable = true, - }; + failureDetails = applicationException.ToTaskFailureDetails(); } } - catch (Exception applicationException) + else { - failureDetails = applicationException.ToTaskFailureDetails(); + if (this.worker.workerOptions.Versioning?.FailureStrategy == DurableTaskWorkerOptions.VersionFailureStrategy.Reject) + { + this.Logger.AbandoningActivityWorkItem(instance.InstanceId, request.Name, request.TaskId, completionToken); + await this.client.AbandonTaskActivityWorkItemAsync( + new P.AbandonActivityTaskRequest + { + CompletionToken = completionToken, + }, + cancellationToken: cancellation); + } + + return; } int outputSizeInBytes = 0; From c4cb43a7872a493ead937d6ba27f7fe6b3a2a6cc Mon Sep 17 00:00:00 2001 From: Hal Spang Date: Tue, 30 Sep 2025 15:04:44 -0700 Subject: [PATCH 2/2] Update changelog Signed-off-by: Hal Spang --- CHANGELOG.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index fe5b175a..a8932d3f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,4 +1,8 @@ # Changelog + +## v1.15.1 (Unreleased) +- Add version check to activities by @halspang in ([#472](https://github.com/microsoft/durabletask-dotnet/pull/472)) + ## v1.15.0 - Abandon workitem if processing workitem failed by @YunchuWang in ([#467](https://github.com/microsoft/durabletask-dotnet/pull/467)) - Extended Sessions for Isolated (Orchestrations) by @sophiatev in ([#449](https://github.com/microsoft/durabletask-dotnet/pull/449))