Update sub-orchestration to use defaultVersion

Signed-off-by: halspang <[email protected]>

Author: halspang

src/WebJobs.Extensions.DurableTask/Bindings/OrchestrationTriggerAttributeBindingProvider.cs

@@ -13,6 +13,7 @@
 using Newtonsoft.Json;
 using Newtonsoft.Json.Linq;
 using static Microsoft.Azure.WebJobs.Extensions.DurableTask.OutOfProcOrchestrationShim;
+using proto = Google.Protobuf.WellKnownTypes;
 
 namespace Microsoft.Azure.WebJobs.Extensions.DurableTask
 {
@@ -172,6 +173,12 @@ public Task<ITriggerData> BindAsync(object? value, ValueBindingContext context)
                         EntityParameters = remoteContext.EntityParameters.ToProtobuf(),
                     };
 
+                    // We only do a null check as an empty string is a valid version.
+                    if (this.config.Options.DefaultVersion != null)
+                    {
+                        orchestratorRequest.Properties.Add("defaultVersion", proto.Value.ForString(this.config.Options.DefaultVersion));
+                    }
+
                     // We convert the binary payload into a base64 string because that seems to be the most commonly supported
                     // format for Azure Functions language workers. Attempts to send unencoded byte[] payloads were unsuccessful.
                     string encodedRequest = ProtobufUtils.Base64Encode(orchestratorRequest);

src/WebJobs.Extensions.DurableTask/Grpc/orchestrator_service.proto

@@ -11,6 +11,7 @@ import "google/protobuf/timestamp.proto";
 import "google/protobuf/duration.proto";
 import "google/protobuf/wrappers.proto";
 import "google/protobuf/empty.proto";
+import "google/protobuf/struct.proto";
 
 message OrchestrationInstance {
     string instanceId = 1;
@@ -75,6 +76,7 @@ message ExecutionStartedEvent {
     google.protobuf.Timestamp scheduledStartTimestamp = 6;
     TraceContext parentTraceContext = 7;
     google.protobuf.StringValue orchestrationSpanID = 8;
+    map<string, string> tags = 9;
 }
 
 message ExecutionCompletedEvent {
@@ -93,6 +95,7 @@ message TaskScheduledEvent {
     google.protobuf.StringValue version = 2;
     google.protobuf.StringValue input = 3;
     TraceContext parentTraceContext = 4;
+    map<string, string> tags = 5;
 }
 
 message TaskCompletedEvent {
@@ -254,6 +257,7 @@ message ScheduleTaskAction {
     string name = 1;
     google.protobuf.StringValue version = 2;
     google.protobuf.StringValue input = 3;
+    map<string, string> tags = 4;
 }
 
 message CreateSubOrchestrationAction {
@@ -288,6 +292,15 @@ message TerminateOrchestrationAction {
     bool recurse = 3;
 }
 
+message SendEntityMessageAction {
+    oneof EntityMessageType {
+        EntityOperationSignaledEvent entityOperationSignaled = 1;
+        EntityOperationCalledEvent entityOperationCalled = 2;
+        EntityLockRequestedEvent entityLockRequested = 3;
+        EntityUnlockSentEvent entityUnlockSent = 4;
+    }
+}
+
 message OrchestratorAction {
     int32 id = 1;
     oneof orchestratorActionType {
@@ -297,6 +310,7 @@ message OrchestratorAction {
         SendEventAction sendEvent = 5;
         CompleteOrchestrationAction completeOrchestration = 6;
         TerminateOrchestrationAction terminateOrchestration = 7;
+        SendEntityMessageAction sendEntityMessage = 8;
     }
 }
 
@@ -307,6 +321,7 @@ message OrchestratorRequest {
     repeated HistoryEvent newEvents = 4;
     OrchestratorEntityParameters entityParameters = 5;
     bool requiresHistoryStreaming = 6;
+    map<string, google.protobuf.Value> properties = 7;
 }
 
 message OrchestratorResponse {
@@ -333,14 +348,8 @@ message CreateInstanceRequest {
 }
 
 message OrchestrationIdReusePolicy {
-    repeated OrchestrationStatus operationStatus = 1;
-    CreateOrchestrationAction action = 2;
-}
-
-enum CreateOrchestrationAction {
-    ERROR = 0;
-    IGNORE = 1;
-    TERMINATE = 2;
+    repeated OrchestrationStatus replaceableStatus = 1;
+    reserved 2;
 }
 
 message CreateInstanceResponse {
@@ -381,6 +390,7 @@ message OrchestrationState {
     google.protobuf.StringValue executionId = 12;
     google.protobuf.Timestamp completedTimestamp = 13;
     google.protobuf.StringValue parentInstanceId = 14;
+    map<string, string> tags = 15;
 }
 
 message RaiseEventRequest {
@@ -457,6 +467,7 @@ message PurgeInstanceFilter {
 
 message PurgeInstancesResponse {
     int32 deletedInstanceCount = 1;
+    google.protobuf.BoolValue isComplete = 2;
 }
 
 message CreateTaskHubRequest {
@@ -551,6 +562,8 @@ message EntityBatchResult {
     repeated OperationAction actions = 2;
     google.protobuf.StringValue entityState = 3;
     TaskFailureDetails failureDetails = 4;
+    string completionToken = 5;
+    repeated OperationInfo operationInfos = 6; // used only with DTS
 }
 
 message EntityRequest {
@@ -573,6 +586,11 @@ message OperationResult {
     }
 }
 
+message OperationInfo {
+    string requestId = 1;
+    OrchestrationInstance responseDestination = 2; // null for signals
+}
+
 message OperationResultSuccess {
     google.protobuf.StringValue result = 1;
 }
@@ -604,6 +622,30 @@ message StartNewOrchestrationAction {
     google.protobuf.Timestamp scheduledTime = 5;
 }
 
+message AbandonActivityTaskRequest {
+    string completionToken = 1;
+}
+
+message AbandonActivityTaskResponse {
+    // Empty.
+}
+
+message AbandonOrchestrationTaskRequest {
+    string completionToken = 1;
+}
+
+message AbandonOrchestrationTaskResponse {
+    // Empty.
+}
+
+message AbandonEntityTaskRequest {
+    string completionToken = 1;
+}
+
+message AbandonEntityTaskResponse {
+    // Empty.
+}
+
 service TaskHubSidecarService {
     // Sends a hello request to the sidecar service.
     rpc Hello(google.protobuf.Empty) returns (google.protobuf.Empty);
@@ -665,6 +707,15 @@ service TaskHubSidecarService {
 
     // clean entity storage
     rpc CleanEntityStorage(CleanEntityStorageRequest) returns (CleanEntityStorageResponse);
+
+    // Abandons a single work item
+    rpc AbandonTaskActivityWorkItem(AbandonActivityTaskRequest) returns (AbandonActivityTaskResponse);
+
+    // Abandon an orchestration work item
+    rpc AbandonTaskOrchestratorWorkItem(AbandonOrchestrationTaskRequest) returns (AbandonOrchestrationTaskResponse);
+
+    // Abandon an entity work item
+    rpc AbandonTaskEntityWorkItem(AbandonEntityTaskRequest) returns (AbandonEntityTaskResponse);
 }
 
 message GetWorkItemsRequest {
@@ -714,4 +765,4 @@ message StreamInstanceHistoryRequest {
 
 message HistoryChunk {
     repeated HistoryEvent events = 1;
-}
+}

src/WebJobs.Extensions.DurableTask/Grpc/versions.txt

@@ -1,2 +1,2 @@
-# The following files were downloaded from branch main at 2025-02-07 22:22:03 UTC
-https://raw.githubusercontent.com/microsoft/durabletask-protobuf/6000187e90d79fba297dfd75d42095abc1462eba/protos/orchestrator_service.proto
+# The following files were downloaded from branch main at 2025-05-29 22:35:16 UTC
+https://raw.githubusercontent.com/microsoft/durabletask-protobuf/d02c540473cf8fe6fc09bcf1fe2ac041b583eb9e/protos/orchestrator_service.proto

src/WebJobs.Extensions.DurableTask/LocalGrpcListener.cs

@@ -225,7 +225,7 @@ public override Task<Empty> Hello(Empty request, ServerCallContext context)
                     ExecutionStartedEvent executionStartedEvent = new ExecutionStartedEvent(-1, request.Input)
                     {
                         Name = request.Name,
-                        Version = !string.IsNullOrEmpty(request.Version) ? request.Version : this.extension.Options.DefaultVersion,
+                        Version = request.Version != null ? request.Version : this.extension.Options.DefaultVersion,
                         OrchestrationInstance = instance,
                         ScheduledStartTime = request.ScheduledStartTimestamp?.ToDateTime(),
                     };

src/WebJobs.Extensions.DurableTask/Options/DurableTaskOptions.cs

@@ -34,7 +34,7 @@ public class DurableTaskOptions
         /// <summary>
         /// The strategy that will be used for matching versions when running an orchestration. See <see cref="VersioningSettings.VersionMatchStrategy"/> for more information.
         /// </summary>
-        public VersioningSettings.VersionMatchStrategy VersionMatchStrategy { get; set; } = VersioningSettings.VersionMatchStrategy.None;
+        public VersioningSettings.VersionMatchStrategy VersionMatchStrategy { get; set; } = VersioningSettings.VersionMatchStrategy.CurrentOrOlder;
 
         /// <summary>
         /// The strategy that will be used if a versioning failure is detected. See <see cref="VersioningSettings.VersionFailureStrategy"/> for more information.

src/Worker.Extensions.DurableTask/FunctionsOrchestrationContext.cs

@@ -89,6 +89,31 @@ public override Task<TResult> CallSubOrchestratorAsync<TResult>(
         TaskName orchestratorName, object? input = null, TaskOptions? options = null)
     {
         this.EnsureLegalAccess();
+        // If we have a default version, add it to the TaskOptions and call the inner context with it.
+        if (this.innerContext.Properties.TryGetValue("defaultVersion", out var propVersion) && propVersion is string defaultVersion)
+        {
+            SubOrchestrationOptions subOptions;
+            if (options is SubOrchestrationOptions subOrchestrationOptions)
+            {
+                subOptions = new SubOrchestrationOptions
+                {
+                    InstanceId = subOrchestrationOptions.InstanceId,
+                    Retry = subOrchestrationOptions.Retry,
+                    Version = subOrchestrationOptions.Version?.Version ?? defaultVersion,
+                };
+            }
+            else
+            {
+                subOptions = new SubOrchestrationOptions
+                {
+                    InstanceId = null, // No instance ID specified, so we use the default one.
+                    Retry = options?.Retry,
+                    Version = defaultVersion
+                };
+            }
+            return this.innerContext.CallSubOrchestratorAsync<TResult>(orchestratorName, input, subOptions);
+        }
+
         return this.innerContext.CallSubOrchestratorAsync<TResult>(orchestratorName, input, options);
     }
 

test/e2e/Apps/BasicDotNetIsolated/VersionedOrchestration.cs

@@ -16,8 +16,40 @@ public static class VersionedOrchestration
     public static async Task<string> RunOrchestrator(
         [OrchestrationTrigger] TaskOrchestrationContext context)
     {
-        ILogger logger = context.CreateReplaySafeLogger(nameof(HelloCities));
-        logger.LogInformation($"Versioned orchestration! Version: {context.Version}");
+        ILogger logger = context.CreateReplaySafeLogger(nameof(VersionedOrchestration));
+        logger.LogInformation($"Versioned orchestration! Version: '{context.Version}'");
+
+        return await context.CallActivityAsync<string>(nameof(SayVersion), context.Version);
+    }
+
+    [Function(nameof(RunWithSubOrchestrator))]
+    public static async Task<string> RunWithSubOrchestrator(
+        [OrchestrationTrigger] TaskOrchestrationContext context,
+        string? subVersion)
+    {
+        ILogger logger = context.CreateReplaySafeLogger(nameof(VersionedOrchestration));
+        logger.LogInformation($"Versioned orchestration! Version: '{context.Version}' Sub Version: '{subVersion}'");
+
+        string subOrchestrationResponse;
+        if (subVersion == null)
+        {
+            subOrchestrationResponse = await context.CallSubOrchestratorAsync<string>(nameof(VersionedSubOrchestration));
+        }
+        else
+        {
+            subOrchestrationResponse = await context.CallSubOrchestratorAsync<string>(nameof(VersionedSubOrchestration), new SubOrchestrationOptions
+            {
+                Version = subVersion,
+            });
+        }
+        return $"Parent Version: '{context.Version}' | Sub {subOrchestrationResponse}";
+    }
+
+    [Function(nameof(VersionedSubOrchestration))]
+    public static async Task<string> VersionedSubOrchestration([OrchestrationTrigger] TaskOrchestrationContext context)
+    {
+        ILogger logger = context.CreateReplaySafeLogger(nameof(VersionedOrchestration));
+        logger.LogInformation($"Versioned sub-orchestration! Version: '{context.Version}'");
 
         return await context.CallActivityAsync<string>(nameof(SayVersion), context.Version);
     }
@@ -27,23 +59,26 @@ public static string SayVersion([ActivityTrigger] string version, FunctionContex
     {
         ILogger logger = executionContext.GetLogger("SayVersion");
         logger.LogInformation("Activity running with version: {name}.", version);
-        return $"Version: {version}";
+        return $"Version: '{version}'";
     }
 
     [Function("OrchestrationVersion_HttpStart")]
     public static async Task<HttpResponseData> HttpStart(
         [HttpTrigger(AuthorizationLevel.Anonymous, "get", "post")] HttpRequestData req,
         [DurableClient] DurableTaskClient client,
         FunctionContext executionContext,
-        string version)
+        string? version)
     {
         ILogger logger = executionContext.GetLogger("VersionedOrchestration_HttpStart");
 
         // Function input comes from the request content.
         string instanceId;
-        if (!string.IsNullOrEmpty(version))
+        if (version != null)
         {
-            instanceId = await client.ScheduleNewOrchestrationInstanceAsync(new TaskName(nameof(VersionedOrchestration), version));
+            instanceId = await client.ScheduleNewOrchestrationInstanceAsync(nameof(VersionedOrchestration), new StartOrchestrationOptions
+            {
+                Version = version,
+            });
         }
         else
         {
@@ -56,4 +91,31 @@ public static async Task<HttpResponseData> HttpStart(
         // See https://learn.microsoft.com/azure/azure-functions/durable/durable-functions-http-api#start-orchestration
         return await client.CreateCheckStatusResponseAsync(req, instanceId);
     }
+
+    [Function("OrchestrationSubVersion_HttpStart")]
+    public static async Task<HttpResponseData> HttpSubStart(
+        [HttpTrigger(AuthorizationLevel.Anonymous, "get", "post")] HttpRequestData req,
+        [DurableClient] DurableTaskClient client,
+        FunctionContext executionContext,
+        string? version)
+    {
+        ILogger logger = executionContext.GetLogger("OrchestrationSubVersion_HttpStart");
+
+        // Function input comes from the request content.
+        string instanceId;
+        if (version != null)
+        {
+            instanceId = await client.ScheduleNewOrchestrationInstanceAsync(nameof(RunWithSubOrchestrator), input: version);
+        }
+        else
+        {
+            instanceId = await client.ScheduleNewOrchestrationInstanceAsync(nameof(RunWithSubOrchestrator));
+        }
+
+        logger.LogInformation("Started orchestration with ID = '{instanceId}' and Version = '{version}'.", instanceId, version);
+
+        // Returns an HTTP 202 response with an instance management payload.
+        // See https://learn.microsoft.com/azure/azure-functions/durable/durable-functions-http-api#start-orchestration
+        return await client.CreateCheckStatusResponseAsync(req, instanceId);
+    }
 }