Update to latest protobuf definitions

Author: cgillum

eng/proto

@@ -17,7 +17,7 @@
   </PropertyGroup>
 
   <PropertyGroup>
-    <VersionPrefix>1.4.0</VersionPrefix>
+    <VersionPrefix>1.5.0</VersionPrefix>
     <VersionSuffix></VersionSuffix>
   </PropertyGroup>
 

eng/targets/Release.props

@@ -11,7 +11,7 @@
     <PackageReference Include="Microsoft.Bcl.AsyncInterfaces" Version="6.0.0" />
     <PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="6.0.0" />
     <PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="6.0.1" />
-    <PackageReference Include="System.Text.Json" Version="6.0.0" />
+    <PackageReference Include="System.Text.Json" Version="6.0.10" />
   </ItemGroup>
 
   <ItemGroup>

src/Abstractions/Abstractions.csproj

@@ -223,18 +223,21 @@ internal static Timestamp ToTimestamp(this DateTime dateTime)
     /// <param name="instanceId">The orchestrator instance ID.</param>
     /// <param name="customStatus">The orchestrator customer status or <c>null</c> if no custom status.</param>
     /// <param name="actions">The orchestrator actions.</param>
+    /// <param name="completionToken">The completion token from the work item.</param>
     /// <returns>The orchestrator response.</returns>
     /// <exception cref="NotSupportedException">When an orchestrator action is unknown.</exception>
     internal static P.OrchestratorResponse ConstructOrchestratorResponse(
         string instanceId,
         string? customStatus,
-        IEnumerable<OrchestratorAction> actions)
+        IEnumerable<OrchestratorAction> actions,
+        string completionToken)
     {
         Check.NotNull(actions);
         var response = new P.OrchestratorResponse
         {
             InstanceId = instanceId,
             CustomStatus = customStatus,
+            CompletionToken = completionToken,
         };
 
         foreach (OrchestratorAction action in actions)
@@ -605,9 +608,12 @@ internal static OrchestrationStatus ToCore(this P.OrchestrationStatus status)
     /// Converts a <see cref="EntityBatchResult" /> to <see cref="P.EntityBatchResult" />.
     /// </summary>
     /// <param name="entityBatchResult">The operation result to convert.</param>
+    /// <param name="completionToken">The completion token from the work item.</param>
     /// <returns>The converted operation result.</returns>
     [return: NotNullIfNotNull(nameof(entityBatchResult))]
-    internal static P.EntityBatchResult? ToEntityBatchResult(this EntityBatchResult? entityBatchResult)
+    internal static P.EntityBatchResult? ToEntityBatchResult(
+        this EntityBatchResult? entityBatchResult,
+        string completionToken)
     {
         if (entityBatchResult == null)
         {
@@ -618,6 +624,7 @@ internal static OrchestrationStatus ToCore(this P.OrchestrationStatus status)
         {
             EntityState = entityBatchResult.EntityState,
             FailureDetails = entityBatchResult.FailureDetails.ToProtobuf(),
+            CompletionToken = completionToken,
         };
 
         foreach (OperationAction action in entityBatchResult.Actions!)

src/Shared/Grpc/ProtoUtils.cs

@@ -83,6 +83,21 @@ public DataConverter DataConverter
     /// </remarks>
     public TimeSpan MaximumTimerInterval { get; set; } = TimeSpan.FromDays(3);
 
+    /// <summary>
+    /// Gets or sets the maximum number of concurrent activity work items that can be processed by the worker.
+    /// </summary>
+    public int MaximumConcurrentActivityWorkItems { get; set; } = 100 * Environment.ProcessorCount;
+
+    /// <summary>
+    /// Gets or sets the maximum number of concurrent orchestration work items that can be processed by the worker.
+    /// </summary>
+    public int MaximumConcurrentOrchestrationWorkItems { get; set; } = 100 * Environment.ProcessorCount;
+
+    /// <summary>
+    /// Gets or sets the maximum number of concurrent entity work items that can be processed by the worker.
+    /// </summary>
+    public int MaximumConcurrentEntityWorkItems { get; set; } = 100 * Environment.ProcessorCount;
+
     /// <summary>
     /// Gets a value indicating whether <see cref="DataConverter" /> was explicitly set or not.
     /// </summary>

src/Worker/Core/DurableTaskWorkerOptions.cs

@@ -11,7 +11,7 @@ The worker is responsible for processing durable task work items.</PackageDescri
   <ItemGroup>
     <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="6.0.0" />
     <PackageReference Include="Microsoft.Extensions.Options" Version="6.0.0" />
-    <PackageReference Include="System.Text.Json" Version="6.0.0" />
+    <PackageReference Include="System.Text.Json" Version="6.0.10" />
   </ItemGroup>
 
   <ItemGroup>

src/Worker/Core/Worker.csproj

@@ -34,7 +34,7 @@ public Processor(GrpcDurableTaskWorker worker, TaskHubSidecarServiceClient sidec
         {
             this.worker = worker;
             this.sidecar = sidecar;
-            this.shimFactory = new DurableTaskShimFactory(this.worker.options, this.worker.loggerFactory);
+            this.shimFactory = new DurableTaskShimFactory(this.worker.grpcOptions, this.worker.loggerFactory);
         }
 
         ILogger Logger => this.worker.logger;
@@ -102,7 +102,7 @@ static OrchestrationRuntimeState BuildRuntimeState(P.OrchestratorRequest request
 
             if (runtimeState.ExecutionStartedEvent == null)
             {
-                // TODO: What's the right way to handle this? Callback to the sidecar with a retryable error request?
+                // TODO: What's the right way to handle this? Callback to the sidecar with a retriable error request?
                 throw new InvalidOperationException("The provided orchestration history was incomplete");
             }
 
@@ -133,8 +133,16 @@ static string GetActionsListForLogging(IReadOnlyList<P.OrchestratorAction> actio
             await this.sidecar!.HelloAsync(EmptyMessage, cancellationToken: cancellation);
             this.Logger.EstablishedWorkItemConnection();
 
+            DurableTaskWorkerOptions workerOptions = this.worker.workerOptions;
+
             // Get the stream for receiving work-items
-            return this.sidecar!.GetWorkItems(new P.GetWorkItemsRequest(), cancellationToken: cancellation);
+            return this.sidecar!.GetWorkItems(
+                new P.GetWorkItemsRequest
+                {
+                    MaxConcurrentActivityWorkItems = workerOptions.MaximumConcurrentActivityWorkItems,
+                    MaxConcurrentOrchestrationWorkItems = workerOptions.MaximumConcurrentOrchestrationWorkItems,
+                },
+                cancellationToken: cancellation);
         }
 
         async Task ProcessWorkItemsAsync(AsyncServerStreamingCall<P.WorkItem> stream, CancellationToken cancellation)
@@ -145,16 +153,25 @@ async Task ProcessWorkItemsAsync(AsyncServerStreamingCall<P.WorkItem> stream, Ca
                 {
                     if (workItem.RequestCase == P.WorkItem.RequestOneofCase.OrchestratorRequest)
                     {
-                        this.RunBackgroundTask(workItem, () => this.OnRunOrchestratorAsync(
-                            workItem.OrchestratorRequest));
+                        this.RunBackgroundTask(
+                            workItem,
+                            () => this.OnRunOrchestratorAsync(workItem.OrchestratorRequest, workItem.CompletionToken));
                     }
                     else if (workItem.RequestCase == P.WorkItem.RequestOneofCase.ActivityRequest)
                     {
-                        this.RunBackgroundTask(workItem, () => this.OnRunActivityAsync(workItem.ActivityRequest));
+                        this.RunBackgroundTask(
+                            workItem,
+                            () => this.OnRunActivityAsync(workItem.ActivityRequest, workItem.CompletionToken));
                     }
                     else if (workItem.RequestCase == P.WorkItem.RequestOneofCase.EntityRequest)
                     {
-                        this.RunBackgroundTask(workItem, () => this.OnRunEntityBatchAsync(workItem.EntityRequest));
+                        this.RunBackgroundTask(
+                            workItem,
+                            () => this.OnRunEntityBatchAsync(workItem.EntityRequest, workItem.CompletionToken));
+                    }
+                    else if (workItem.RequestCase == P.WorkItem.RequestOneofCase.HealthPing)
+                    {
+                        // No-op
                     }
                     else
                     {
@@ -188,7 +205,7 @@ void RunBackgroundTask(P.WorkItem? workItem, Func<Task> handler)
             });
         }
 
-        async Task OnRunOrchestratorAsync(P.OrchestratorRequest request)
+        async Task OnRunOrchestratorAsync(P.OrchestratorRequest request, string completionToken)
         {
             OrchestratorExecutionResult? result = null;
             P.TaskFailureDetails? failureDetails = null;
@@ -248,14 +265,16 @@ async Task OnRunOrchestratorAsync(P.OrchestratorRequest request)
                 response = ProtoUtils.ConstructOrchestratorResponse(
                     request.InstanceId,
                     result.CustomStatus,
-                    result.Actions);
+                    result.Actions,
+                    completionToken);
             }
             else
             {
                 // This is the case for failures that happened *outside* the orchestrator executor
                 response = new P.OrchestratorResponse
                 {
                     InstanceId = request.InstanceId,
+                    CompletionToken = completionToken,
                     Actions =
                     {
                         new P.OrchestratorAction
@@ -279,7 +298,7 @@ async Task OnRunOrchestratorAsync(P.OrchestratorRequest request)
             await this.sidecar.CompleteOrchestratorTaskAsync(response);
         }
 
-        async Task OnRunActivityAsync(P.ActivityRequest request)
+        async Task OnRunActivityAsync(P.ActivityRequest request, string completionToken)
         {
             OrchestrationInstance instance = request.OrchestrationInstance.ToCore();
             string rawInput = request.Input;
@@ -336,12 +355,13 @@ async Task OnRunActivityAsync(P.ActivityRequest request)
                 TaskId = request.TaskId,
                 Result = output,
                 FailureDetails = failureDetails,
+                CompletionToken = completionToken,
             };
 
             await this.sidecar.CompleteActivityTaskAsync(response);
         }
 
-        async Task OnRunEntityBatchAsync(P.EntityBatchRequest request)
+        async Task OnRunEntityBatchAsync(P.EntityBatchRequest request, string completionToken)
         {
             var coreEntityId = DTCore.Entities.EntityId.FromString(request.InstanceId);
             EntityId entityId = new(coreEntityId.Name, coreEntityId.Key);
@@ -400,7 +420,7 @@ async Task OnRunEntityBatchAsync(P.EntityBatchRequest request)
             }
 
             // convert the result to protobuf format and send it back
-            P.EntityBatchResult response = batchResult.ToEntityBatchResult();
+            P.EntityBatchResult response = batchResult.ToEntityBatchResult(completionToken);
             await this.sidecar.CompleteEntityTaskAsync(response);
         }
     }

src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs

@@ -12,7 +12,8 @@ namespace Microsoft.DurableTask.Worker.Grpc;
 /// </summary>
 sealed partial class GrpcDurableTaskWorker : DurableTaskWorker
 {
-    readonly GrpcDurableTaskWorkerOptions options;
+    readonly GrpcDurableTaskWorkerOptions grpcOptions;
+    readonly DurableTaskWorkerOptions workerOptions;
     readonly IServiceProvider services;
     readonly ILoggerFactory loggerFactory;
     readonly ILogger logger;
@@ -22,18 +23,21 @@ sealed partial class GrpcDurableTaskWorker : DurableTaskWorker
     /// </summary>
     /// <param name="name">The name of the worker.</param>
     /// <param name="factory">The task factory.</param>
-    /// <param name="options">The gRPC worker options.</param>
+    /// <param name="grpcOptions">The gRPC-specific worker options.</param>
+    /// <param name="workerOptions">The generic worker options.</param>
     /// <param name="services">The service provider.</param>
     /// <param name="loggerFactory">The logger.</param>
     public GrpcDurableTaskWorker(
         string name,
         IDurableTaskFactory factory,
-        IOptionsMonitor<GrpcDurableTaskWorkerOptions> options,
+        IOptionsMonitor<GrpcDurableTaskWorkerOptions> grpcOptions,
+        IOptionsMonitor<DurableTaskWorkerOptions> workerOptions,
         IServiceProvider services,
         ILoggerFactory loggerFactory)
         : base(name, factory)
     {
-        this.options = Check.NotNull(options).Get(name);
+        this.grpcOptions = Check.NotNull(grpcOptions).Get(name);
+        this.workerOptions = Check.NotNull(workerOptions).Get(name);
         this.services = Check.NotNull(services);
         this.loggerFactory = Check.NotNull(loggerFactory);
         this.logger = loggerFactory.CreateLogger("Microsoft.DurableTask"); // TODO: use better category name.
@@ -73,19 +77,19 @@ static GrpcChannel GetChannel(string? address)
 
     AsyncDisposable GetCallInvoker(out CallInvoker callInvoker)
     {
-        if (this.options.Channel is GrpcChannel c)
+        if (this.grpcOptions.Channel is GrpcChannel c)
         {
             callInvoker = c.CreateCallInvoker();
             return default;
         }
 
-        if (this.options.CallInvoker is CallInvoker invoker)
+        if (this.grpcOptions.CallInvoker is CallInvoker invoker)
         {
             callInvoker = invoker;
             return default;
         }
 
-        c = GetChannel(this.options.Address);
+        c = GetChannel(this.grpcOptions.Address);
         callInvoker = c.CreateCallInvoker();
         return new AsyncDisposable(() => new(c.ShutdownAsync()));
     }

src/Worker/Grpc/GrpcDurableTaskWorker.cs

@@ -69,7 +69,7 @@ public static async Task<string> LoadAndRunAsync(
         TaskEntity entity = factory.CreateEntity(entityName, implementation, id);
         EntityBatchResult result = await entity.ExecuteOperationBatchAsync(batch);
 
-        P.EntityBatchResult response = result.ToEntityBatchResult();
+        P.EntityBatchResult response = result.ToEntityBatchResult(completionToken: string.Empty /* doesn't apply */);
         byte[] responseBytes = response.ToByteArray();
         return Convert.ToBase64String(responseBytes);
     }

src/Worker/Grpc/GrpcEntityRunner.cs

@@ -115,7 +115,8 @@ public static string LoadAndRun(
         P.OrchestratorResponse response = ProtoUtils.ConstructOrchestratorResponse(
             request.InstanceId,
             result.CustomStatus,
-            result.Actions);
+            result.Actions,
+            completionToken: string.Empty /* doesn't apply */);
         byte[] responseBytes = response.ToByteArray();
         return Convert.ToBase64String(responseBytes);
     }