Expand worker capacibilities

Author: YunchuWang

src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs

@@ -8,6 +8,7 @@
 using Microsoft.DurableTask.Worker.Grpc;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Options;
+using P = Microsoft.DurableTask.Protobuf;
 
 namespace Microsoft.DurableTask;
 
@@ -77,6 +78,8 @@ static IDurableTaskWorkerBuilder UseExternalizedPayloadsCore(IDurableTaskWorkerB
                     throw new ArgumentException(
                         "Channel or CallInvoker must be provided to use Azure Blob Payload Externalization feature");
                 }
+
+                opt.Capabilities.Add(P.WorkerCapability.LargePayloads);
             });
 
         return builder;

src/Grpc/orchestrator_service.proto

@@ -692,14 +692,14 @@ message AbandonEntityTaskResponse {
 }
 
 message SkipGracefulOrchestrationTerminationsRequest {
-    InstanceBatch instanceBatch = 1;
-    google.protobuf.StringValue reason = 2;
+    InstanceBatch instanceBatch = 1;
+    google.protobuf.StringValue reason = 2;
 }
 
 message SkipGracefulOrchestrationTerminationsResponse {
     // Those instances which could not be terminated because they had locked entities at the time of this termination call,
     // are already in a terminal state (completed, failed, terminated, etc.), are not orchestrations, or do not exist (i.e. have been purged)
-    repeated string unterminatedInstanceIds = 1;
+    repeated string unterminatedInstanceIds = 1;
 }
 
 service TaskHubSidecarService {
@@ -797,6 +797,10 @@ enum WorkerCapability {
     // When set, the service may return work items without any history events as an optimization.
     // It is strongly recommended that all SDKs support this capability.
     WORKER_CAPABILITY_HISTORY_STREAMING = 1;
+
+    WORKER_CAPABILITY_SCHEDULED_TASKS = 2;
+
+    WORKER_CAPABILITY_LARGE_PAYLOADS = 3;
 }
 
 message WorkItem {

src/ScheduledTasks/Extension/DurableTaskWorkerBuilderExtensions.cs

@@ -1,6 +1,9 @@
 // Copyright (c) Microsoft Corporation.
 // Licensed under the MIT License.
 using Microsoft.DurableTask.Worker;
+using Microsoft.DurableTask.Worker.Grpc;
+using Microsoft.Extensions.DependencyInjection;
+using P = Microsoft.DurableTask.Protobuf;
 
 namespace Microsoft.DurableTask.ScheduledTasks;
 
@@ -20,5 +23,13 @@ public static void UseScheduledTasks(this IDurableTaskWorkerBuilder builder)
             r.AddEntity<Schedule>();
             r.AddOrchestrator<ExecuteScheduleOperationOrchestrator>();
         });
+
+        // Register the feature for gRPC workers
+        builder.Services
+            .AddOptions<GrpcDurableTaskWorkerOptions>(builder.Name)
+            .PostConfigure(opt =>
+            {
+                opt.Capabilities.Add(P.WorkerCapability.ScheduledTasks);
+            });
     }
 }

src/ScheduledTasks/ScheduledTasks.csproj

@@ -11,6 +11,7 @@
     <ProjectReference Include="../Abstractions/Abstractions.csproj" />
     <ProjectReference Include="../Client/Core/Client.csproj" />
     <ProjectReference Include="../Worker/Core/Worker.csproj" />
+    <ProjectReference Include="../Worker/Grpc/Worker.Grpc.csproj" />
   </ItemGroup>
 
   <ItemGroup>

src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs

@@ -254,7 +254,7 @@ async ValueTask<OrchestrationRuntimeState> BuildRuntimeStateAsync(
                         workerOptions.Concurrency.MaximumConcurrentOrchestrationWorkItems,
                     MaxConcurrentEntityWorkItems =
                         workerOptions.Concurrency.MaximumConcurrentEntityWorkItems,
-                    Capabilities = { P.WorkerCapability.HistoryStreaming },
+                    Capabilities = { this.worker.grpcOptions.Capabilities },
                 },
                 cancellationToken: cancellation);
         }

src/Worker/Grpc/GrpcDurableTaskWorkerOptions.cs

@@ -1,6 +1,8 @@
 // Copyright (c) Microsoft Corporation.
 // Licensed under the MIT License.
 
+using P = Microsoft.DurableTask.Protobuf;
+
 namespace Microsoft.DurableTask.Worker.Grpc;
 
 /// <summary>
@@ -23,6 +25,12 @@ public sealed class GrpcDurableTaskWorkerOptions : DurableTaskWorkerOptions
     /// </summary>
     public CallInvoker? CallInvoker { get; set; }
 
+    /// <summary>
+    /// Gets the collection of capabilities enabled on this worker.
+    /// Capabilities are announced to the backend on connection.
+    /// </summary>
+    public HashSet<P.WorkerCapability> Capabilities { get; } = new() { P.WorkerCapability.HistoryStreaming };
+
     /// <summary>
     /// Gets the internal protocol options. These are used to control backend-dependent features.
     /// </summary>