update versioning

Author: YunchuWang

packages/durabletask-js/src/index.ts

@@ -4,6 +4,7 @@
 // Client and Worker
 export { TaskHubGrpcClient, TaskHubGrpcClientOptions, MetadataGenerator } from "./client/client";
 export { TaskHubGrpcWorker, TaskHubGrpcWorkerOptions } from "./worker/task-hub-grpc-worker";
+export { VersioningOptions, VersionMatchStrategy, VersionFailureStrategy } from "./worker/versioning-options";
 
 // Contexts
 export { OrchestrationContext } from "./task/context/orchestration-context";

packages/durabletask-js/src/worker/task-hub-grpc-worker.ts

@@ -18,6 +18,8 @@ import { ActivityExecutor } from "./activity-executor";
 import { StringValue } from "google-protobuf/google/protobuf/wrappers_pb";
 import { Logger, ConsoleLogger } from "../types/logger.type";
 import { ExponentialBackoff, sleep, withTimeout } from "../utils/backoff.util";
+import { VersioningOptions, VersionMatchStrategy, VersionFailureStrategy } from "./versioning-options";
+import { compareVersions } from "../utils/versioning.util";
 
 /** Default timeout in milliseconds for graceful shutdown. */
 const DEFAULT_SHUTDOWN_TIMEOUT_MS = 30000;
@@ -40,6 +42,8 @@ export interface TaskHubGrpcWorkerOptions {
   logger?: Logger;
   /** Optional timeout in milliseconds for graceful shutdown. Defaults to 30000. */
   shutdownTimeoutMs?: number;
+  /** Optional versioning options for filtering orchestrations by version. */
+  versioning?: VersioningOptions;
 }
 
 export class TaskHubGrpcWorker {
@@ -57,6 +61,7 @@ export class TaskHubGrpcWorker {
   private _pendingWorkItems: Set<Promise<void>>;
   private _shutdownTimeoutMs: number;
   private _backoff: ExponentialBackoff;
+  private _versioning?: VersioningOptions;
 
   /**
    * Creates a new TaskHubGrpcWorker instance.
@@ -103,6 +108,7 @@ export class TaskHubGrpcWorker {
     let resolvedMetadataGenerator: MetadataGenerator | undefined;
     let resolvedLogger: Logger | undefined;
     let resolvedShutdownTimeoutMs: number | undefined;
+    let resolvedVersioning: VersioningOptions | undefined;
 
     if (typeof hostAddressOrOptions === "object" && hostAddressOrOptions !== null) {
       // Options object constructor
@@ -113,6 +119,7 @@ export class TaskHubGrpcWorker {
       resolvedMetadataGenerator = hostAddressOrOptions.metadataGenerator;
       resolvedLogger = hostAddressOrOptions.logger;
       resolvedShutdownTimeoutMs = hostAddressOrOptions.shutdownTimeoutMs;
+      resolvedVersioning = hostAddressOrOptions.versioning;
     } else {
       // Deprecated positional parameters constructor
       resolvedHostAddress = hostAddressOrOptions;
@@ -142,6 +149,7 @@ export class TaskHubGrpcWorker {
       maxDelayMs: 30000,
       multiplier: 2,
     });
+    this._versioning = resolvedVersioning;
   }
 
   /**
@@ -400,6 +408,74 @@ export class TaskHubGrpcWorker {
     await sleep(1000);
   }
 
+  /**
+   * Result of version compatibility check.
+   */
+  private _checkVersionCompatibility(req: pb.OrchestratorRequest): {
+    compatible: boolean;
+    shouldFail: boolean;
+    orchestrationVersion?: string;
+  } {
+    // If no versioning options configured or match strategy is None, always compatible
+    if (!this._versioning || this._versioning.matchStrategy === VersionMatchStrategy.None) {
+      return { compatible: true, shouldFail: false };
+    }
+
+    // Extract orchestration version from ExecutionStarted event
+    const orchestrationVersion = this._getOrchestrationVersion(req);
+    const workerVersion = this._versioning.version;
+
+    // If worker version is not set, process all
+    if (!workerVersion) {
+      return { compatible: true, shouldFail: false };
+    }
+
+    let compatible = false;
+
+    switch (this._versioning.matchStrategy) {
+      case VersionMatchStrategy.Strict:
+        // Only process if versions match exactly
+        compatible = orchestrationVersion === workerVersion;
+        break;
+
+      case VersionMatchStrategy.CurrentOrOlder:
+        // Process if orchestration version is current or older
+        if (!orchestrationVersion) {
+          // Empty orchestration version is considered older
+          compatible = true;
+        } else {
+          compatible = compareVersions(orchestrationVersion, workerVersion) <= 0;
+        }
+        break;
+
+      default:
+        compatible = true;
+    }
+
+    if (!compatible) {
+      const shouldFail = this._versioning.failureStrategy === VersionFailureStrategy.Fail;
+      return { compatible: false, shouldFail, orchestrationVersion };
+    }
+
+    return { compatible: true, shouldFail: false };
+  }
+
+  /**
+   * Extracts the orchestration version from the ExecutionStarted event in the request.
+   */
+  private _getOrchestrationVersion(req: pb.OrchestratorRequest): string | undefined {
+    // Look for ExecutionStarted event in both past and new events
+    const allEvents = [...req.getPasteventsList(), ...req.getNeweventsList()];
+
+    for (const event of allEvents) {
+      if (event.hasExecutionstarted()) {
+        return event.getExecutionstarted()?.getVersion()?.getValue();
+      }
+    }
+
+    return undefined;
+  }
+
   /**
    * Executes an orchestrator request and tracks it as a pending work item.
    */
@@ -429,6 +505,55 @@ export class TaskHubGrpcWorker {
       throw new Error(`Could not execute the orchestrator as the instanceId was not provided (${instanceId})`);
     }
 
+    // Check version compatibility if versioning is enabled
+    const versionCheckResult = this._checkVersionCompatibility(req);
+    if (!versionCheckResult.compatible) {
+      if (versionCheckResult.shouldFail) {
+        // Fail the orchestration with version mismatch error
+        this._logger.warn(
+          `Version mismatch for instance '${instanceId}': orchestration version '${versionCheckResult.orchestrationVersion}' does not match worker version '${this._versioning?.version}'. Failing orchestration.`,
+        );
+
+        const failureDetails = pbh.newFailureDetails(
+          new Error(`Version mismatch: orchestration version '${versionCheckResult.orchestrationVersion}' is not compatible with worker version '${this._versioning?.version}'`),
+        );
+
+        const actions = [
+          pbh.newCompleteOrchestrationAction(
+            -1,
+            pb.OrchestrationStatus.ORCHESTRATION_STATUS_FAILED,
+            failureDetails?.toString(),
+          ),
+        ];
+
+        const res = new pb.OrchestratorResponse();
+        res.setInstanceid(instanceId);
+        res.setCompletiontoken(completionToken);
+        res.setActionsList(actions);
+
+        try {
+          await callWithMetadata(stub.completeOrchestratorTask.bind(stub), res, this._metadataGenerator);
+        } catch (e: any) {
+          this._logger.error(`An error occurred while trying to complete instance '${instanceId}': ${e?.message}`);
+        }
+        return;
+      } else {
+        // Reject the work item - explicitly abandon it so it can be picked up by another worker
+        this._logger.info(
+          `Version mismatch for instance '${instanceId}': orchestration version '${versionCheckResult.orchestrationVersion}' does not match worker version '${this._versioning?.version}'. Abandoning work item.`,
+        );
+
+        try {
+          const abandonRequest = new pb.AbandonOrchestrationTaskRequest();
+          abandonRequest.setCompletiontoken(completionToken);
+          await callWithMetadata(stub.abandonTaskOrchestratorWorkItem.bind(stub), abandonRequest, this._metadataGenerator);
+        } catch (e: any) {
+          this._logger.error(`An error occurred while trying to abandon work item for instance '${instanceId}': ${e?.message}`);
+        }
+        return;
+      }
+    }
+
     let res;
 
     try {

packages/durabletask-js/src/worker/versioning-options.ts

@@ -0,0 +1,62 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+/**
+ * Strategy for matching orchestration versions when processing work items.
+ */
+export enum VersionMatchStrategy {
+  /**
+   * No version matching - process all orchestrations regardless of version.
+   */
+  None = 0,
+
+  /**
+   * Only process orchestrations that exactly match the worker's version.
+   */
+  Strict = 1,
+
+  /**
+   * Process orchestrations with the current version or older versions.
+   * Uses semantic versioning comparison.
+   */
+  CurrentOrOlder = 2,
+}
+
+/**
+ * Strategy for handling version mismatches when processing work items.
+ */
+export enum VersionFailureStrategy {
+  /**
+   * Reject the work item and let it be picked up by another worker.
+   * The orchestration will be retried by a compatible worker.
+   */
+  Reject = 0,
+
+  /**
+   * Fail the orchestration with a version mismatch error.
+   * This will mark the orchestration as failed.
+   */
+  Fail = 1,
+}
+
+/**
+ * Options for configuring version-based filtering of orchestrations.
+ */
+export interface VersioningOptions {
+  /**
+   * The version of the worker. This is used for version matching when processing orchestrations.
+   */
+  version?: string;
+
+  /**
+   * The strategy for matching orchestration versions.
+   * @default VersionMatchStrategy.None
+   */
+  matchStrategy?: VersionMatchStrategy;
+
+  /**
+   * The strategy for handling version mismatches.
+   * @default VersionFailureStrategy.Reject
+   */
+  failureStrategy?: VersionFailureStrategy;
+}

packages/durabletask-js/test/versioning-options.spec.ts

@@ -0,0 +1,82 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+import { VersionMatchStrategy, VersionFailureStrategy, VersioningOptions } from "../src/worker/versioning-options";
+
+describe("VersioningOptions", () => {
+  describe("VersionMatchStrategy enum", () => {
+    it("should have None = 0", () => {
+      expect(VersionMatchStrategy.None).toBe(0);
+    });
+
+    it("should have Strict = 1", () => {
+      expect(VersionMatchStrategy.Strict).toBe(1);
+    });
+
+    it("should have CurrentOrOlder = 2", () => {
+      expect(VersionMatchStrategy.CurrentOrOlder).toBe(2);
+    });
+  });
+
+  describe("VersionFailureStrategy enum", () => {
+    it("should have Reject = 0", () => {
+      expect(VersionFailureStrategy.Reject).toBe(0);
+    });
+
+    it("should have Fail = 1", () => {
+      expect(VersionFailureStrategy.Fail).toBe(1);
+    });
+  });
+
+  describe("VersioningOptions interface", () => {
+    it("should allow creating options with all properties", () => {
+      const options: VersioningOptions = {
+        version: "1.0.0",
+        matchStrategy: VersionMatchStrategy.Strict,
+        failureStrategy: VersionFailureStrategy.Fail,
+      };
+
+      expect(options.version).toBe("1.0.0");
+      expect(options.matchStrategy).toBe(VersionMatchStrategy.Strict);
+      expect(options.failureStrategy).toBe(VersionFailureStrategy.Fail);
+    });
+
+    it("should allow creating options with only version", () => {
+      const options: VersioningOptions = {
+        version: "2.0.0",
+      };
+
+      expect(options.version).toBe("2.0.0");
+      expect(options.matchStrategy).toBeUndefined();
+      expect(options.failureStrategy).toBeUndefined();
+    });
+
+    it("should allow creating empty options", () => {
+      const options: VersioningOptions = {};
+
+      expect(options.version).toBeUndefined();
+      expect(options.matchStrategy).toBeUndefined();
+      expect(options.failureStrategy).toBeUndefined();
+    });
+
+    it("should allow CurrentOrOlder match strategy", () => {
+      const options: VersioningOptions = {
+        version: "3.0.0",
+        matchStrategy: VersionMatchStrategy.CurrentOrOlder,
+        failureStrategy: VersionFailureStrategy.Reject,
+      };
+
+      expect(options.matchStrategy).toBe(VersionMatchStrategy.CurrentOrOlder);
+      expect(options.failureStrategy).toBe(VersionFailureStrategy.Reject);
+    });
+
+    it("should allow None match strategy (process all versions)", () => {
+      const options: VersioningOptions = {
+        version: "1.0.0",
+        matchStrategy: VersionMatchStrategy.None,
+      };
+
+      expect(options.matchStrategy).toBe(VersionMatchStrategy.None);
+    });
+  });
+});