update

nytian · nytian · commit 3d4e4853e81f · 2025-10-02T10:39:11.000-07:00
diff --git 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/eng/targets/Release.props b/eng/targets/Release.props
@@ -17,7 +17,7 @@
   </PropertyGroup>
 
   <PropertyGroup>
-    <VersionPrefix>1.15.0</VersionPrefix>
+    <VersionPrefix>1.15.1</VersionPrefix>
     <VersionSuffix></VersionSuffix>
   </PropertyGroup>
 
diff --git a/src/Abstractions/TaskFailureDetails.cs b/src/Abstractions/TaskFailureDetails.cs
@@ -16,7 +16,7 @@ namespace Microsoft.DurableTask;
 /// <param name="StackTrace">The stack trace of the failure.</param>
 /// <param name="InnerFailure">The inner cause of the task failure.</param>
 /// <param name="Properties">Additional properties associated with the exception.</param>
-public record TaskFailureDetails(string ErrorType, string ErrorMessage, string? StackTrace, TaskFailureDetails? InnerFailure, IDictionary<string, object>? Properties)
+public record TaskFailureDetails(string ErrorType, string ErrorMessage, string? StackTrace, TaskFailureDetails? InnerFailure, IDictionary<string, object?>? Properties)
 {
     Type? loadedExceptionType;
 
diff --git a/src/Shared/Grpc/ProtoUtils.cs b/src/Shared/Grpc/ProtoUtils.cs
diff --git a/src/Worker/Core/IExceptionPropertiesProvider.cs b/src/Worker/Core/IExceptionPropertiesProvider.cs
@@ -6,9 +6,6 @@ namespace Microsoft.DurableTask.Worker;
 /// <summary>
 /// Provides custom exception property inclusion rules for enriching FailureDetails.
 /// </summary>
-/// <remarks>
-/// Implementations should be thread-safe. The worker may call this for many exceptions concurrently.
-/// </remarks>
 public interface IExceptionPropertiesProvider
 {
         /// <summary>
diff --git a/src/Worker/Core/Shims/DurableTaskShimFactory.cs b/src/Worker/Core/Shims/DurableTaskShimFactory.cs
@@ -26,7 +26,6 @@ public class DurableTaskShimFactory
     /// </summary>
     /// <param name="options">The data converter.</param>
     /// <param name="loggerFactory">The logger factory.</param>
-    /// <param name="exceptionPropertiesProvider">Custom provider used to extract exception properties for inclusion in the failure details.</param>
     public DurableTaskShimFactory(
         DurableTaskWorkerOptions? options = null, ILoggerFactory? loggerFactory = null)
     {
diff --git a/src/Worker/Core/Shims/TaskOrchestrationContextWrapper.cs b/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(),
diff --git a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs
@@ -33,7 +33,7 @@ class Processor
         readonly TaskHubSidecarServiceClient client;
         readonly DurableTaskShimFactory shimFactory;
         readonly GrpcDurableTaskWorkerOptions.InternalOptions internalOptions;
-        readonly DTCore.IExceptionPropertiesProvider exceptionPropertiesProvider;
+        readonly DTCore.IExceptionPropertiesProvider? exceptionPropertiesProvider;
         [Obsolete("Experimental")]
         readonly IOrchestrationFilter? orchestrationFilter;
 
@@ -44,7 +44,7 @@ public Processor(GrpcDurableTaskWorker worker, TaskHubSidecarServiceClient clien
             this.shimFactory = new DurableTaskShimFactory(this.worker.grpcOptions, this.worker.loggerFactory);
             this.internalOptions = this.worker.grpcOptions.Internal;
             this.orchestrationFilter = orchestrationFilter;
-            this.exceptionPropertiesProvider = exceptionPropertiesProvider != null
+            this.exceptionPropertiesProvider = exceptionPropertiesProvider is not null
                 ? new ExceptionPropertiesProviderAdapter(exceptionPropertiesProvider)
                 : null;
         }
@@ -764,16 +764,60 @@ 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);
             innerContext.ExceptionPropertiesProvider = this.exceptionPropertiesProvider;
 
             TaskName name = new(request.Name);
             string? output = null;
-            P.TaskFailureDetails? failureDetails = null;
+
+            failureDetails = EvaluateOrchestrationVersioning(this.worker.workerOptions.Versioning, request.Version, out bool versioningFailed);
+            if (!versioningFailed)
+            {
+                try
+                {
+                    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,
+                        };
+                    }
+                }
+                catch (Exception applicationException)
+                {
+                    failureDetails = applicationException.ToTaskFailureDetails(this.exceptionPropertiesProvider);
+                }
+            }
+            else
+            {
+                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;
+            }
+
             try
             {
                 await using AsyncServiceScope scope = this.worker.services.CreateAsyncScope();
diff --git a/test/Grpc.IntegrationTests/GrpcSidecar/Grpc/ProtobufUtils.cs b/test/Grpc.IntegrationTests/GrpcSidecar/Grpc/ProtobufUtils.cs
@@ -359,7 +359,7 @@ public static string Base64Encode(IMessage message)
             failureDetails.StackTrace,
             GetFailureDetails(failureDetails.InnerFailure),
             failureDetails.IsNonRetriable,
-            ConvertProperties(failureDetails.Properties));
+            ConvertMapToDictionary(failureDetails.Properties));
     }
 
     internal static Proto.TaskFailureDetails? GetFailureDetails(FailureDetails? failureDetails)
@@ -368,7 +368,7 @@ public static string Base64Encode(IMessage message)
         {
             return null;
         }
-        
+
         var taskFailureDetails = new Proto.TaskFailureDetails
         {
             ErrorType = failureDetails.ErrorType,
@@ -454,7 +454,7 @@ internal static Proto.PurgeInstancesResponse CreatePurgeInstancesResponse(PurgeR
     }
 
     /// <summary>
-    /// Converts a Dictionary<string, object?> into a MapField<string, Value>.
+    /// Converts a IDictionary<string, object?> into a MapField<string, Value>.
     /// Supports nested dictionaries and lists.
     /// </summary>
     public static MapField<string, Value> ConvertDictionaryToMapField(IDictionary<string, object?> dict)
@@ -470,15 +470,15 @@ public static MapField<string, Value> ConvertDictionaryToMapField(IDictionary<st
     }
 
     /// <summary>
-    /// 
+    /// Converts a MapField<string, Value> into a IDictionary<string, object?>.
     /// </summary>
     /// <param name="properties"></param>
     /// <returns></returns>
-    public static IDictionary<string, object> ConvertProperties(MapField<string, Value> properties)
+    public static IDictionary<string, object?> ConvertMapToDictionary(MapField<string, Value> properties)
     {
         return properties.ToDictionary(
             kvp => kvp.Key,
-            kvp => ConvertValue(kvp.Value)
+            kvp => ConvertValueToObject(kvp.Value)
         );
     }
 
@@ -487,7 +487,7 @@ public static IDictionary<string, object> ConvertProperties(MapField<string, Val
     /// </summary>
     /// <param name="obj">The object to convert.</param>
     /// <returns>The converted protobuf Value.</returns>
-    private static Value ConvertObjectToValue(object? obj)
+    static Value ConvertObjectToValue(object? obj)
     {
         return obj switch
         {
@@ -510,7 +510,7 @@ private static Value ConvertObjectToValue(object? obj)
         };
     }
 
-    private static object ConvertValue(Value value)
+    static object ConvertValueToObject(Value value)
     {
         switch (value.KindCase)
         {
@@ -521,9 +521,9 @@ private static object ConvertValue(Value value)
             case Value.KindOneofCase.BoolValue:
                 return value.BoolValue;
             case Value.KindOneofCase.StructValue:
-                return value.StructValue.Fields.ToDictionary(f => f.Key, f => ConvertValue(f.Value));
+                return value.StructValue.Fields.ToDictionary(f => f.Key, f => ConvertValueToObject(f.Value));
             case Value.KindOneofCase.ListValue:
-                return value.ListValue.Values.Select(ConvertValue).ToList();
+                return value.ListValue.Values.Select(ConvertValueToObject).ToList();
             case Value.KindOneofCase.NullValue:
                 return null!;
             default:
diff --git a/test/Grpc.IntegrationTests/OrchestrationErrorHandling.cs b/test/Grpc.IntegrationTests/OrchestrationErrorHandling.cs
@@ -633,15 +633,15 @@ static Exception MakeException(Type exceptionType, string message)
     }
 
     /// <summary>
-    /// Tests that custom exception properties are included in FailureDetails when using a custom IExceptionPropertiesProvider.
+    /// Tests that exception properties are included in FailureDetails when using a custom IExceptionPropertiesProvider.
     /// </summary>
     [Fact]
     public async Task CustomExceptionPropertiesInFailureDetails()
     {
         TaskName orchestratorName = "OrchestrationWithCustomException";
         TaskName activityName = "BusinessActivity";
 
-        // Use local function definitions
+        // Register activity functions that will throw a exception.
         async Task MyOrchestrationImpl(TaskOrchestrationContext ctx) =>
             await ctx.CallActivityAsync(activityName);
 

Original file line number	Diff line number	Diff line change
`@@ -16,7 +16,7 @@ namespace Microsoft.DurableTask;`
`16`	`16`	`/// <param name="StackTrace">The stack trace of the failure.</param>`
`17`	`17`	`/// <param name="InnerFailure">The inner cause of the task failure.</param>`
`18`	`18`	`/// <param name="Properties">Additional properties associated with the exception.</param>`
`19`		`-public record TaskFailureDetails(string ErrorType, string ErrorMessage, string? StackTrace, TaskFailureDetails? InnerFailure, IDictionary<string, object>? Properties)`
	`19`	`+public record TaskFailureDetails(string ErrorType, string ErrorMessage, string? StackTrace, TaskFailureDetails? InnerFailure, IDictionary<string, object?>? Properties)`
`20`	`20`	`{`
`21`	`21`	`Type? loadedExceptionType;`
`22`	`22`
Original file line number	Diff line number	Diff line change
`@@ -26,7 +26,6 @@ public class DurableTaskShimFactory`
`26`	`26`	`/// </summary>`
`27`	`27`	`/// <param name="options">The data converter.</param>`
`28`	`28`	`/// <param name="loggerFactory">The logger factory.</param>`
`29`		`- /// <param name="exceptionPropertiesProvider">Custom provider used to extract exception properties for inclusion in the failure details.</param>`
`30`	`29`	`public DurableTaskShimFactory(`
`31`	`30`	`DurableTaskWorkerOptions? options = null, ILoggerFactory? loggerFactory = null)`
`32`	`31`	`{`
Original file line number	Diff line number	Diff line change
`@@ -359,7 +359,7 @@ public static string Base64Encode(IMessage message)`
`359`	`359`	`failureDetails.StackTrace,`
`360`	`360`	`GetFailureDetails(failureDetails.InnerFailure),`
`361`	`361`	`failureDetails.IsNonRetriable,`
`362`		`- ConvertProperties(failureDetails.Properties));`
	`362`	`+ ConvertMapToDictionary(failureDetails.Properties));`
`363`	`363`	`}`
`364`	`364`
`365`	`365`	`internal static Proto.TaskFailureDetails? GetFailureDetails(FailureDetails? failureDetails)`
`@@ -368,7 +368,7 @@ public static string Base64Encode(IMessage message)`
`368`	`368`	`{`
`369`	`369`	`return null;`
`370`	`370`	`}`
`371`		`-`
	`371`	`+`
`372`	`372`	`var taskFailureDetails = new Proto.TaskFailureDetails`
`373`	`373`	`{`
`374`	`374`	`ErrorType = failureDetails.ErrorType,`
`@@ -454,7 +454,7 @@ internal static Proto.PurgeInstancesResponse CreatePurgeInstancesResponse(PurgeR`
`454`	`454`	`}`
`455`	`455`
`456`	`456`	`/// <summary>`
`457`		`- /// Converts a Dictionary<string, object?> into a MapField<string, Value>.`
	`457`	`+ /// Converts a IDictionary<string, object?> into a MapField<string, Value>.`
`458`	`458`	`/// Supports nested dictionaries and lists.`
`459`	`459`	`/// </summary>`
`460`	`460`	`public static MapField<string, Value> ConvertDictionaryToMapField(IDictionary<string, object?> dict)`
`@@ -470,15 +470,15 @@ public static MapField<string, Value> ConvertDictionaryToMapField(IDictionary<st`
`470`	`470`	`}`
`471`	`471`
`472`	`472`	`/// <summary>`
`473`		`- ///`
	`473`	`+ /// Converts a MapField<string, Value> into a IDictionary<string, object?>.`
`474`	`474`	`/// </summary>`
`475`	`475`	`/// <param name="properties"></param>`
`476`	`476`	`/// <returns></returns>`
`477`		`- public static IDictionary<string, object> ConvertProperties(MapField<string, Value> properties)`
	`477`	`+ public static IDictionary<string, object?> ConvertMapToDictionary(MapField<string, Value> properties)`
`478`	`478`	`{`
`479`	`479`	`return properties.ToDictionary(`
`480`	`480`	`kvp => kvp.Key,`
`481`		`- kvp => ConvertValue(kvp.Value)`
	`481`	`+ kvp => ConvertValueToObject(kvp.Value)`
`482`	`482`	`);`
`483`	`483`	`}`
`484`	`484`
`@@ -487,7 +487,7 @@ public static IDictionary<string, object> ConvertProperties(MapField<string, Val`
`487`	`487`	`/// </summary>`
`488`	`488`	`/// <param name="obj">The object to convert.</param>`
`489`	`489`	`/// <returns>The converted protobuf Value.</returns>`
`490`		`- private static Value ConvertObjectToValue(object? obj)`
	`490`	`+ static Value ConvertObjectToValue(object? obj)`
`491`	`491`	`{`
`492`	`492`	`return obj switch`
`493`	`493`	`{`
`@@ -510,7 +510,7 @@ private static Value ConvertObjectToValue(object? obj)`
`510`	`510`	`};`
`511`	`511`	`}`
`512`	`512`
`513`		`- private static object ConvertValue(Value value)`
	`513`	`+ static object ConvertValueToObject(Value value)`
`514`	`514`	`{`
`515`	`515`	`switch (value.KindCase)`
`516`	`516`	`{`
`@@ -521,9 +521,9 @@ private static object ConvertValue(Value value)`
`521`	`521`	`case Value.KindOneofCase.BoolValue:`
`522`	`522`	`return value.BoolValue;`
`523`	`523`	`case Value.KindOneofCase.StructValue:`
`524`		`- return value.StructValue.Fields.ToDictionary(f => f.Key, f => ConvertValue(f.Value));`
	`524`	`+ return value.StructValue.Fields.ToDictionary(f => f.Key, f => ConvertValueToObject(f.Value));`
`525`	`525`	`case Value.KindOneofCase.ListValue:`
`526`		`- return value.ListValue.Values.Select(ConvertValue).ToList();`
	`526`	`+ return value.ListValue.Values.Select(ConvertValueToObject).ToList();`
`527`	`527`	`case Value.KindOneofCase.NullValue:`
`528`	`528`	`return null!;`
`529`	`529`	`default:`