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