Introduce WorkItemFilters into worker flow

src/Grpc/orchestrator_service.proto

@@ -822,6 +822,7 @@ message GetWorkItemsRequest {
     int32 maxConcurrentEntityWorkItems = 3;
 
     repeated WorkerCapability capabilities = 10;
+    WorkItemFilters workItemFilters = 11;
 }
 
 enum WorkerCapability {
@@ -844,6 +845,26 @@ enum WorkerCapability {
     WORKER_CAPABILITY_LARGE_PAYLOADS = 3;
 }
 
+message WorkItemFilters {
+    repeated OrchestrationFilter orchestrations = 1;
+    repeated ActivityFilter activities = 2;
+    repeated EntityFilter entities = 3;
+}
+
+message OrchestrationFilter {
+    string name = 1;
+    repeated string versions = 2;
+}
+
+message ActivityFilter {
+    string name = 1;
+    repeated string versions = 2;
+}
+
+message EntityFilter {
+    string name = 1;
+}
+
 message WorkItem {
     oneof request {
         OrchestratorRequest orchestratorRequest = 1;

src/Worker/Core/DependencyInjection/DurableTaskWorkerBuilderExtensions.cs

@@ -78,7 +78,7 @@
         where TTarget : DurableTaskWorker
         where TOptions : DurableTaskWorkerOptions
     {
         builder.UseBuildTarget(typeof(TTarget));
         builder.Services.AddOptions<TOptions>(builder.Name)
             .PostConfigure<IOptionsMonitor<DurableTaskWorkerOptions>>((options, baseOptions) =>
             {
@@ -137,4 +137,32 @@
         builder.Services.AddSingleton(filter);
         return builder;
     }
+
+    /// <summary>
+    /// Adds <see cref="DurableTaskWorkerWorkItemFilters"/> to the specified <see cref="IDurableTaskWorkerBuilder"/>.
+    /// </summary>
+    /// <param name="builder">The builder to set the builder target for.</param>
+    /// <param name="workItemFilters">The instance of a <see cref="DurableTaskWorkerWorkItemFilters"/> to use.</param>
+    /// <returns>The same <see cref="IDurableTaskWorkerBuilder"/> instance, allowing for method chaining.</returns>
+    /// <remarks>If this is called without specified filters, the filters will be constructed from the registered orchestrations, activities, and entities.</remarks>
+    public static IDurableTaskWorkerBuilder UseWorkItemFilters(this IDurableTaskWorkerBuilder builder, DurableTaskWorkerWorkItemFilters? workItemFilters = null)
+    {
+        Check.NotNull(builder);
+        if (workItemFilters != null)
+        {
+            builder.Services.AddSingleton(workItemFilters);
+        }
+        else
+        {
+            // Auto-generate the filters from registered orchestrations, activities, and entities.
+            builder.Services.AddSingleton(provider =>
+            {
+                DurableTaskRegistry registry = provider.GetRequiredService<IOptionsMonitor<DurableTaskRegistry>>().Get(builder.Name);
+                DurableTaskWorkerOptions? options = provider.GetOptions<DurableTaskWorkerOptions>(builder.Name);
+                return DurableTaskWorkerWorkItemFilters.FromDurableTaskRegistry(registry, options);
+            });
+        }
+
+        return builder;
+    }
 }

src/Worker/Core/DurableTaskWorkerWorkItemFilters.cs

@@ -0,0 +1,99 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+namespace Microsoft.DurableTask.Worker;
+
+/// <summary>
+/// A class that represents work item filters for a Durable Task Worker. These filters are passed to the backend
+/// and only work items matching the filters will be processed by the worker. If no filters are provided,
+/// the worker will process all work items.
+/// </summary>
+public class DurableTaskWorkerWorkItemFilters
+{
+    /// <summary>
+    /// Gets or initializes the orchestration filters.
+    /// </summary>
+    public IReadOnlyList<OrchestrationFilter> Orchestrations { get; init; } = [];
+
+    /// <summary>
+    /// Gets or initializes the activity filters.
+    /// </summary>
+    public IReadOnlyList<ActivityFilter> Activities { get; init; } = [];
+
+    /// <summary>
+    /// Gets or initializes the entity filters.
+    /// </summary>
+    public IReadOnlyList<EntityFilter> Entities { get; init; } = [];
+
+    /// <summary>
+    /// Creates a new instance of the <see cref="DurableTaskWorkerWorkItemFilters"/> class.
+    /// </summary>
+    /// <param name="registry"><see cref="DurableTaskRegistry"/> to construct the filter from.</param>
+    /// <param name="workerOptions"><see cref="DurableTaskWorkerOptions"/> that optionally provides versioning information.</param>
+    /// <returns>A new instance of <see cref="DurableTaskWorkerWorkItemFilters"/> constructed from the provided registry.</returns>
+    internal static DurableTaskWorkerWorkItemFilters FromDurableTaskRegistry(DurableTaskRegistry registry, DurableTaskWorkerOptions? workerOptions)
+    {
+        // TODO: Support multiple versions per orchestration/activity. For now, grab the worker version from the options.
+        return new DurableTaskWorkerWorkItemFilters
+        {
+            Orchestrations = registry.Orchestrators.Select(orchestration => new OrchestrationFilter
+            {
+                Name = orchestration.Key,
+                Versions = workerOptions?.Versioning != null ? [workerOptions.Versioning.DefaultVersion] : [],
+            }).ToList(),
+            Activities = registry.Activities.Select(activity => new ActivityFilter
+            {
+                Name = activity.Key,
+                Versions = workerOptions?.Versioning != null ? [workerOptions.Versioning.DefaultVersion] : [],
+            }).ToList(),
+            Entities = registry.Entities.Select(entity => new EntityFilter
+            {
+                // Entity names are normalized to lowercase in the backend.
+                Name = entity.Key.ToString().ToLowerInvariant(),
+            }).ToList(),
+        };
+    }
+
+    /// <summary>
+    /// Struct specifying an orchestration filter.
+    /// </summary>
+    public struct OrchestrationFilter
+    {
+        /// <summary>
+        /// Gets or initializes the name of the orchestration to filter.
+        /// </summary>
+        public string Name { get; init; }
+
+        /// <summary>
+        /// Gets or initializes the versions of the orchestration to filter.
+        /// </summary>
+        public IReadOnlyList<string> Versions { get; init; }
+    }
+
+    /// <summary>
+    /// Struct specifying an activity filter.
+    /// </summary>
+    public struct ActivityFilter
+    {
+        /// <summary>
+        /// Gets or initializes the name of the activity to filter.
+        /// </summary>
+        public string Name { get; init; }
+
+        /// <summary>
+        /// Gets or initializes the versions of the activity to filter.
+        /// </summary>
+        public IReadOnlyList<string> Versions { get; init; }
+    }
+
+    /// <summary>
+    /// Struct specifying an entity filter.
+    /// </summary>
+    public struct EntityFilter
+    {
+        /// <summary>
+        /// Gets or initializes the name of the entity to filter.
+        /// </summary>
+        public string Name { get; init; }
+    }
+}

src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs

@@ -10,6 +10,7 @@
 using Microsoft.DurableTask.Abstractions;
 using Microsoft.DurableTask.Entities;
 using Microsoft.DurableTask.Tracing;
+using Microsoft.DurableTask.Worker.Grpc.Internal;
 using Microsoft.DurableTask.Worker.Shims;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
@@ -255,6 +256,7 @@ async ValueTask<OrchestrationRuntimeState> BuildRuntimeStateAsync(
                     MaxConcurrentEntityWorkItems =
                         workerOptions.Concurrency.MaximumConcurrentEntityWorkItems,
                     Capabilities = { this.worker.grpcOptions.Capabilities },
+                    WorkItemFilters = this.worker.workItemFilters?.ToGrpcWorkItemFilters(),
                 },
                 cancellationToken: cancellation);
         }

src/Worker/Grpc/GrpcDurableTaskWorker.cs

@@ -18,6 +18,7 @@ sealed partial class GrpcDurableTaskWorker : DurableTaskWorker
     readonly ILoggerFactory loggerFactory;
     readonly ILogger logger;
     readonly IOrchestrationFilter? orchestrationFilter;
+    readonly DurableTaskWorkerWorkItemFilters? workItemFilters;
 
     /// <summary>
     /// Initializes a new instance of the <see cref="GrpcDurableTaskWorker" /> class.
@@ -30,6 +31,7 @@ sealed partial class GrpcDurableTaskWorker : DurableTaskWorker
     /// <param name="loggerFactory">The logger.</param>
     /// <param name="orchestrationFilter">The optional <see cref="IOrchestrationFilter"/> used to filter orchestration execution.</param>
     /// <param name="exceptionPropertiesProvider">The custom exception properties provider that help build failure details.</param>
+    /// <param name="workItemFilters">The optional <see cref="DurableTaskWorkerWorkItemFilters"/> used to filter work items in the backend.</param>
     public GrpcDurableTaskWorker(
         string name,
         IDurableTaskFactory factory,
@@ -38,7 +40,8 @@ public GrpcDurableTaskWorker(
         IServiceProvider services,
         ILoggerFactory loggerFactory,
         IOrchestrationFilter? orchestrationFilter = null,
-        IExceptionPropertiesProvider? exceptionPropertiesProvider = null)
+        IExceptionPropertiesProvider? exceptionPropertiesProvider = null,
+        DurableTaskWorkerWorkItemFilters? workItemFilters = null)
         : base(name, factory)
     {
         this.grpcOptions = Check.NotNull(grpcOptions).Get(name);
@@ -48,6 +51,7 @@ public GrpcDurableTaskWorker(
         this.logger = CreateLogger(loggerFactory, this.workerOptions);
         this.orchestrationFilter = orchestrationFilter;
         this.ExceptionPropertiesProvider = exceptionPropertiesProvider;
+        this.workItemFilters = workItemFilters;
     }
 
     /// <inheritdoc />

src/Worker/Grpc/Internal/DurableTaskWorkerWorkItemFiltersExtension.cs

@@ -0,0 +1,53 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+using P = Microsoft.DurableTask.Protobuf;
+
+namespace Microsoft.DurableTask.Worker.Grpc.Internal;
+
+/// <summary>
+/// Extension for <see cref="DurableTaskWorkerWorkItemFilters"/> to convert to gRPC types.
+/// </summary>
+public static class DurableTaskWorkerWorkItemFiltersExtension
+{
+    /// <summary>
+    /// Converts a <see cref="DurableTaskWorkerWorkItemFilters"/> to a gRPC <see cref="P.WorkItemFilters"/>.
+    /// </summary>
+    /// <param name="workItemFilter">The <see cref="DurableTaskWorkerWorkItemFilters"/> to convert.</param>
+    /// <returns>A gRPC <see cref="P.WorkItemFilters"/>.</returns>
+    public static P.WorkItemFilters ToGrpcWorkItemFilters(this DurableTaskWorkerWorkItemFilters workItemFilter)
+    {
+        Check.NotNull(workItemFilter);
+        var grpcWorkItemFilters = new P.WorkItemFilters();
+        foreach (var orchestrationFilter in workItemFilter.Orchestrations)
+        {
+            var grpcOrchestrationFilter = new P.OrchestrationFilter
+            {
+                Name = orchestrationFilter.Name,
+            };
+            grpcOrchestrationFilter.Versions.AddRange(orchestrationFilter.Versions);
+            grpcWorkItemFilters.Orchestrations.Add(grpcOrchestrationFilter);
+        }
+
+        foreach (var activityFilter in workItemFilter.Activities)
+        {
+            var grpcActivityFilter = new P.ActivityFilter
+            {
+                Name = activityFilter.Name,
+            };
+            grpcActivityFilter.Versions.AddRange(activityFilter.Versions);
+            grpcWorkItemFilters.Activities.Add(grpcActivityFilter);
+        }
+
+        foreach (var entityFilter in workItemFilter.Entities)
+        {
+            var grpcEntityFilter = new P.EntityFilter
+            {
+                Name = entityFilter.Name,
+            };
+            grpcWorkItemFilters.Entities.Add(grpcEntityFilter);
+        }
+
+        return grpcWorkItemFilters;
+    }
+}