Add version check to activities

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

src/Worker/Core/Shims/TaskOrchestrationContextWrapper.cs

@@ -154,7 +154,7 @@ public override async Task<T> CallActivityAsync<T>(
             {
                 return await this.innerContext.ScheduleTask<T>(
                     name.Name,
-                    name.Version,
+                    this.innerContext.Version,
                     options: ScheduleTaskOptions.CreateBuilder()
                         .WithRetryOptions(policy.ToDurableTaskCoreRetryOptions())
                         .WithTags(tags)
@@ -166,7 +166,7 @@ public override async Task<T> CallActivityAsync<T>(
                 return await this.InvokeWithCustomRetryHandler(
                     () => this.innerContext.ScheduleTask<T>(
                         name.Name,
-                        name.Version,
+                        this.innerContext.Version,
                         options: ScheduleTaskOptions.CreateBuilder()
                             .WithTags(tags)
                             .Build(),
@@ -179,7 +179,7 @@ public override async Task<T> CallActivityAsync<T>(
             {
                 return await this.innerContext.ScheduleTask<T>(
                     name.Name,
-                    name.Version,
+                    this.innerContext.Version,
                     options: ScheduleTaskOptions.CreateBuilder()
                         .WithTags(tags)
                         .Build(),

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;