feat: Implement work item filters (#168)

Author: YunchuWang

examples/work-item-filters/index.ts

@@ -0,0 +1,150 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+/**
+ * This example demonstrates Work Item Filters for Durable Task workers.
+ *
+ * Work Item Filters allow a worker to tell the sidecar which orchestrations,
+ * activities, and entities it is configured to handle. The sidecar then only
+ * dispatches matching work items to that worker, enabling efficient routing.
+ *
+ * Key concepts demonstrated:
+ * - Auto-generated filters from the worker's registry (default behavior)
+ * - Explicit filters via useWorkItemFilters()
+ *
+ * This example runs against:
+ *   DTS Emulator:
+ *     docker run --name dts-emulator -i -p 8080:8080 -d --rm mcr.microsoft.com/dts/dts-emulator:latest
+ *   Then:
+ *     npx ts-node --swc examples/work-item-filters/index.ts
+ */
+
+import {
+  ActivityContext,
+  OrchestrationContext,
+  TOrchestrator,
+  WorkItemFilters,
+} from "@microsoft/durabletask-js";
+import {
+  DurableTaskAzureManagedClientBuilder,
+  DurableTaskAzureManagedWorkerBuilder,
+} from "@microsoft/durabletask-js-azuremanaged";
+
+const endpoint = process.env.ENDPOINT || "localhost:8080";
+const taskHub = process.env.TASKHUB || "default";
+
+// ============================================================================
+// Step 1: Define activities and orchestrators
+// ============================================================================
+
+const greet = async (_: ActivityContext, name: string): Promise<string> => {
+  return `Hello, ${name}!`;
+};
+
+const add = async (_: ActivityContext, input: { a: number; b: number }): Promise<number> => {
+  return input.a + input.b;
+};
+
+const greetingOrchestrator: TOrchestrator = async function* (
+  ctx: OrchestrationContext,
+  name: string,
+): Promise<any> {
+  const result = yield ctx.callActivity(greet, name);
+  return result;
+};
+
+const mathOrchestrator: TOrchestrator = async function* (
+  ctx: OrchestrationContext,
+  input: { a: number; b: number },
+): Promise<any> {
+  const result = yield ctx.callActivity(add, input);
+  return result;
+};
+
+// ============================================================================
+// Step 2: Demonstrate different work item filter configurations
+// ============================================================================
+
+async function runWithAutoGeneratedFilters() {
+  console.log("\n=== Scenario 1: Auto-Generated Filters (Default) ===");
+  console.log("The worker auto-generates filters from its registered orchestrators and activities.");
+  console.log("Only matching work items will be dispatched to this worker.\n");
+
+  const client = new DurableTaskAzureManagedClientBuilder()
+    .endpoint(endpoint, taskHub, null)
+    .build();
+
+  // No explicit filters — they are auto-generated from addOrchestrator/addActivity
+  const worker = new DurableTaskAzureManagedWorkerBuilder()
+    .endpoint(endpoint, taskHub, null)
+    .addOrchestrator(greetingOrchestrator)
+    .addActivity(greet)
+    .build();
+
+  await worker.start();
+  console.log("Worker started with auto-generated filters for: greetingOrchestrator, greet");
+
+  const id = await client.scheduleNewOrchestration(greetingOrchestrator, "Auto-Filters");
+  console.log(`Scheduled orchestration: ${id}`);
+
+  const state = await client.waitForOrchestrationCompletion(id, undefined, 30);
+  console.log(`Result: ${state?.serializedOutput}`);
+
+  await worker.stop();
+  await client.stop();
+}
+
+async function runWithExplicitFilters() {
+  console.log("\n=== Scenario 2: Explicit Filters ===");
+  console.log("The worker uses explicitly provided filters instead of auto-generating them.");
+  console.log("This is useful when you want fine-grained control over which work items to accept.\n");
+
+  const client = new DurableTaskAzureManagedClientBuilder()
+    .endpoint(endpoint, taskHub, null)
+    .build();
+
+  // Provide explicit filters — these override auto-generation
+  const filters: WorkItemFilters = {
+    orchestrations: [{ name: "mathOrchestrator" }],
+    activities: [{ name: "add" }],
+  };
+
+  const worker = new DurableTaskAzureManagedWorkerBuilder()
+    .endpoint(endpoint, taskHub, null)
+    .addOrchestrator(mathOrchestrator)
+    .addActivity(add)
+    .useWorkItemFilters(filters)
+    .build();
+
+  await worker.start();
+  console.log("Worker started with explicit filters for: mathOrchestrator, add");
+
+  const id = await client.scheduleNewOrchestration(mathOrchestrator, { a: 17, b: 25 });
+  console.log(`Scheduled orchestration: ${id}`);
+
+  const state = await client.waitForOrchestrationCompletion(id, undefined, 30);
+  console.log(`Result: ${state?.serializedOutput}`);
+
+  await worker.stop();
+  await client.stop();
+}
+
+// ============================================================================
+// Step 3: Run all scenarios
+// ============================================================================
+
+(async () => {
+  console.log(`Connecting to DTS emulator at ${endpoint}, taskHub: ${taskHub}`);
+
+  try {
+    await runWithAutoGeneratedFilters();
+    await runWithExplicitFilters();
+
+    console.log("\n=== All scenarios completed successfully! ===");
+  } catch (error) {
+    console.error("Error:", error);
+    process.exit(1);
+  }
+
+  process.exit(0);
+})();

examples/work-item-filters/package.json

@@ -0,0 +1,17 @@
+{
+  "name": "work-item-filters-example",
+  "version": "1.0.0",
+  "description": "Example demonstrating work item filters for Durable Task workers",
+  "private": true,
+  "scripts": {
+    "start": "ts-node --swc index.ts",
+    "start:emulator": "ENDPOINT=localhost:8080 TASKHUB=default ts-node --swc index.ts"
+  },
+  "dependencies": {
+    "@microsoft/durabletask-js": "workspace:*",
+    "@microsoft/durabletask-js-azuremanaged": "workspace:*"
+  },
+  "engines": {
+    "node": ">=22.0.0"
+  }
+}

internal/protocol/SOURCE_COMMIT

@@ -1 +1 @@
-026329c53fe6363985655857b9ca848ec7238bd2
+1caadbd7ecfdf5f2309acbeac28a3e36d16aa156

internal/protocol/protos/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;

package-lock.json

@@ -13,6 +13,7 @@ import {
   Logger,
   ConsoleLogger,
   VersioningOptions,
+  WorkItemFilters,
 } from "@microsoft/durabletask-js";
 
 /**
@@ -27,6 +28,7 @@ export class DurableTaskAzureManagedWorkerBuilder {
   private _logger: Logger = new ConsoleLogger();
   private _shutdownTimeoutMs?: number;
   private _versioning?: VersioningOptions;
+  private _workItemFilters?: WorkItemFilters | "auto";
 
   /**
    * Creates a new instance of DurableTaskAzureManagedWorkerBuilder.
@@ -220,6 +222,21 @@ export class DurableTaskAzureManagedWorkerBuilder {
     return this;
   }
 
+  /**
+   * Enables work item filters for the worker.
+   * When called without arguments, filters are auto-generated from the registered
+   * orchestrations, activities, and entities.
+   * When called with a WorkItemFilters object, those specific filters are used.
+   * By default (when not called), no filters are sent and the worker processes all work items.
+   *
+   * @param filters Optional explicit filters. Omit to auto-generate from registry.
+   * @returns This builder instance.
+   */
+  useWorkItemFilters(filters?: WorkItemFilters): DurableTaskAzureManagedWorkerBuilder {
+    this._workItemFilters = filters ?? "auto";
+    return this;
+  }
+
   /**
    * Builds and returns a configured TaskHubGrpcWorker.
    *
@@ -251,6 +268,7 @@ export class DurableTaskAzureManagedWorkerBuilder {
       logger: this._logger,
       shutdownTimeoutMs: this._shutdownTimeoutMs,
       versioning: this._versioning,
+      workItemFilters: this._workItemFilters,
     });
 
     // Register all orchestrators