feat: Add versioning support and replay-safe logging recursive terminating/purging

Author: YunchuWang

packages/durabletask-js/src/client/client.ts

@@ -17,6 +17,8 @@ import { OrchestrationStatus, toProtobuf, fromProtobuf } from "../orchestration/
 import { TimeoutError } from "../exception/timeout-error";
 import { PurgeResult } from "../orchestration/orchestration-purge-result";
 import { PurgeInstanceCriteria } from "../orchestration/orchestration-purge-criteria";
+import { PurgeInstanceOptions } from "../orchestration/orchestration-purge-options";
+import { TerminateInstanceOptions } from "../orchestration/orchestration-terminate-options";
 import { callWithMetadata, MetadataGenerator } from "../utils/grpc-helper.util";
 import { OrchestrationQuery, ListInstanceIdsOptions, DEFAULT_PAGE_SIZE } from "../orchestration/orchestration-query";
 import { Page, AsyncPageable, createAsyncPageable } from "../orchestration/page";
@@ -177,6 +179,10 @@ export class TaskHubGrpcClient {
       typeof instanceIdOrOptions === "string" || instanceIdOrOptions === undefined
         ? undefined
         : instanceIdOrOptions.tags;
+    const version =
+      typeof instanceIdOrOptions === "string" || instanceIdOrOptions === undefined
+        ? undefined
+        : instanceIdOrOptions.version;
 
     const req = new pb.CreateInstanceRequest();
     req.setName(name);
@@ -191,9 +197,15 @@ export class TaskHubGrpcClient {
     req.setInput(i);
     req.setScheduledstarttimestamp(ts);
 
+    if (version) {
+      const v = new StringValue();
+      v.setValue(version);
+      req.setVersion(v);
+    }
+
     populateTagsMap(req.getTagsMap(), tags);
 
-    this._logger.info(`Starting new ${name} instance with ID = ${req.getInstanceid()}`);
+    this._logger.info(`Starting new ${name} instance with ID = ${req.getInstanceid()}${version ? ` (version: ${version})` : ''}`);
 
     const res = await callWithMetadata<pb.CreateInstanceRequest, pb.CreateInstanceResponse>(
       this._stub.startInstance.bind(this._stub),
@@ -364,18 +376,38 @@ export class TaskHubGrpcClient {
    * Terminates the orchestrator associated with the provided instance id.
    *
    * @param {string} instanceId - orchestrator instance id to terminate.
-   * @param {any} output - The optional output to set for the terminated orchestrator instance.
+   * @param {any | TerminateInstanceOptions} outputOrOptions - The optional output to set for the terminated orchestrator instance,
+   *        or an options object that can include both output and recursive termination settings.
    */
-  async terminateOrchestration(instanceId: string, output: any = null): Promise<void> {
+  async terminateOrchestration(
+    instanceId: string,
+    outputOrOptions: any | TerminateInstanceOptions = null,
+  ): Promise<void> {
     const req = new pb.TerminateRequest();
     req.setInstanceid(instanceId);
 
+    let output: any = null;
+    let recursive = false;
+
+    // Check if outputOrOptions is a TerminateInstanceOptions object
+    if (
+      outputOrOptions !== null &&
+      typeof outputOrOptions === 'object' &&
+      ('recursive' in outputOrOptions || 'output' in outputOrOptions)
+    ) {
+      output = outputOrOptions.output ?? null;
+      recursive = outputOrOptions.recursive ?? false;
+    } else {
+      output = outputOrOptions;
+    }
+
     const i = new StringValue();
     i.setValue(JSON.stringify(output));
 
     req.setOutput(i);
+    req.setRecursive(recursive);
 
-    this._logger.info(`Terminating '${instanceId}'`);
+    this._logger.info(`Terminating '${instanceId}'${recursive ? ' (recursive)' : ''}`);
 
     await callWithMetadata<pb.TerminateRequest, pb.TerminateResponse>(
       this._stub.terminateInstance.bind(this._stub),
@@ -537,16 +569,21 @@ export class TaskHubGrpcClient {
    *
    * @param value - The unique ID of the orchestration instance to purge or orchestration instance filter criteria used
    * to determine which instances to purge.
+   * @param options - Optional options to control the purge behavior, such as recursive purging of sub-orchestrations.
    * @returns A Promise that resolves to a {@link PurgeResult} or `undefined` if the purge operation was not successful.
    */
-  async purgeOrchestration(value: string | PurgeInstanceCriteria): Promise<PurgeResult | undefined> {
+  async purgeOrchestration(
+    value: string | PurgeInstanceCriteria,
+    options?: PurgeInstanceOptions,
+  ): Promise<PurgeResult | undefined> {
     let res;
     if (typeof value === `string`) {
       const instanceId = value;
       const req = new pb.PurgeInstancesRequest();
       req.setInstanceid(instanceId);
+      req.setRecursive(options?.recursive ?? false);
 
-      this._logger.info(`Purging Instance '${instanceId}'`);
+      this._logger.info(`Purging Instance '${instanceId}'${options?.recursive ? ' (recursive)' : ''}`);
 
       res = await callWithMetadata<pb.PurgeInstancesRequest, pb.PurgeInstancesResponse>(
         this._stub.purgeInstances.bind(this._stub),
@@ -574,9 +611,10 @@ export class TaskHubGrpcClient {
         filter.addRuntimestatus(toProtobuf(status));
       }
       req.setPurgeinstancefilter(filter);
+      req.setRecursive(options?.recursive ?? false);
       const timeout = purgeInstanceCriteria.getTimeout();
 
-      this._logger.info("Purging Instance using purging criteria");
+      this._logger.info(`Purging Instances using purging criteria${options?.recursive ? ' (recursive)' : ''}`);
 
       const callPromise = callWithMetadata<pb.PurgeInstancesRequest, pb.PurgeInstancesResponse>(
         this._stub.purgeInstances.bind(this._stub),

packages/durabletask-js/src/index.ts

@@ -11,6 +11,8 @@ export { ActivityContext } from "./task/context/activity-context";
 
 // Orchestration types and utilities
 export { PurgeInstanceCriteria } from "./orchestration/orchestration-purge-criteria";
+export { PurgeInstanceOptions } from "./orchestration/orchestration-purge-options";
+export { TerminateInstanceOptions } from "./orchestration/orchestration-terminate-options";
 export { OrchestrationStatus } from "./orchestration/enum/orchestration-status.enum";
 export { OrchestrationState } from "./orchestration/orchestration-state";
 
@@ -81,3 +83,7 @@ export { TOutput } from "./types/output.type";
 
 // Logger
 export { Logger, ConsoleLogger, NoOpLogger } from "./types/logger.type";
+export { ReplaySafeLogger, ReplayContext } from "./types/replay-safe-logger";
+
+// Versioning utilities
+export { compareVersions } from "./utils/versioning.util";

packages/durabletask-js/src/orchestration/orchestration-purge-options.ts

@@ -0,0 +1,18 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+/**
+ * Options for purging orchestration instances.
+ */
+export interface PurgeInstanceOptions {
+  /**
+   * Whether to recursively purge sub-orchestrations as well.
+   * When true, all child orchestrations spawned by the target orchestration
+   * will also be purged.
+   *
+   * Note: Recursive purging may not be supported by all backend implementations.
+   *
+   * @default false
+   */
+  recursive?: boolean;
+}

packages/durabletask-js/src/orchestration/orchestration-terminate-options.ts

@@ -0,0 +1,21 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+/**
+ * Options for terminating orchestration instances.
+ */
+export interface TerminateInstanceOptions {
+  /**
+   * Whether to recursively terminate sub-orchestrations as well.
+   * When true, all child orchestrations spawned by the target orchestration
+   * will also be terminated.
+   *
+   * @default false
+   */
+  recursive?: boolean;
+
+  /**
+   * The optional output to set for the terminated orchestrator instance.
+   */
+  output?: any;
+}

packages/durabletask-js/src/task/context/orchestration-context.ts

@@ -3,6 +3,8 @@
 
 import { TActivity } from "../../types/activity.type";
 import { TOrchestrator } from "../../types/orchestrator.type";
+import { Logger } from "../../types/logger.type";
+import { ReplaySafeLogger } from "../../types/replay-safe-logger";
 import { TaskOptions, SubOrchestrationOptions } from "../options";
 import { Task } from "../task";
 
@@ -38,6 +40,17 @@ export abstract class OrchestrationContext {
    */
   abstract get isReplaying(): boolean;
 
+  /**
+   * Gets the version of the current orchestration instance.
+   *
+   * The version is set when the orchestration instance is created via the client's
+   * scheduleNewOrchestration method using StartOrchestrationOptions.version.
+   * If no version was specified, this returns an empty string.
+   *
+   * @returns {string} The version of the current orchestration instance.
+   */
+  abstract get version(): string;
+
   /**
    * Create a timer task that will fire at a specified time.
    *
@@ -128,4 +141,28 @@ export abstract class OrchestrationContext {
    * @returns {string} A new deterministic UUID string.
    */
   abstract newGuid(): string;
+
+  /**
+   * Creates a replay-safe logger that only writes logs when the orchestrator is not replaying.
+   *
+   * During orchestration replay, history events are re-processed to rebuild state.
+   * This can cause duplicate log entries if not handled properly. The returned logger
+   * wraps the provided logger and automatically suppresses log output during replay,
+   * ensuring that logs are only written once when the orchestration is making forward progress.
+   *
+   * @param {Logger} logger The underlying logger to wrap.
+   * @returns {Logger} A replay-safe logger instance.
+   *
+   * @example
+   * ```typescript
+   * const orchestrator: TOrchestrator = async function* (ctx, input) {
+   *   const logger = ctx.createReplaySafeLogger(myLogger);
+   *   logger.info("This will only be logged once, not during replay");
+   *   yield ctx.callActivity(myActivity, input);
+   * };
+   * ```
+   */
+  createReplaySafeLogger(logger: Logger): Logger {
+    return new ReplaySafeLogger(this, logger);
+  }
 }

packages/durabletask-js/src/task/options/task-options.ts

@@ -16,6 +16,11 @@ export interface TaskOptions {
    * The tags to associate with the task.
    */
   tags?: Record<string, string>;
+  /**
+   * The version of the task (activity) to execute.
+   * When specified, only workers that handle this version will process the task.
+   */
+  version?: string;
 }
 
 /**
@@ -48,6 +53,12 @@ export interface StartOrchestrationOptions {
    * The tags to associate with the orchestration instance.
    */
   tags?: Record<string, string>;
+  /**
+   * The version of the orchestration to execute.
+   * This version is stored with the orchestration instance and can be accessed
+   * via the OrchestrationContext.version property.
+   */
+  version?: string;
 }
 
 /**

packages/durabletask-js/src/types/replay-safe-logger.ts

@@ -0,0 +1,96 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+import { Logger } from "./logger.type";
+
+/**
+ * Interface representing a context that can determine if replay is occurring.
+ * This is used by ReplaySafeLogger to check if logging should be suppressed.
+ */
+export interface ReplayContext {
+  /**
+   * Whether the orchestrator is currently replaying from history.
+   */
+  readonly isReplaying: boolean;
+}
+
+/**
+ * A logger wrapper that only logs when the orchestration is not replaying.
+ *
+ * During orchestration replay, history events are re-processed to rebuild state.
+ * This can cause duplicate log entries if not handled properly. The ReplaySafeLogger
+ * wraps an existing logger and automatically suppresses log output during replay,
+ * ensuring that logs are only written once when the orchestration is making forward progress.
+ *
+ * @example
+ * ```typescript
+ * // Inside an orchestrator function:
+ * const logger = ctx.createReplaySafeLogger(myLogger);
+ * logger.info("This will only be logged once, not during replay");
+ * ```
+ */
+export class ReplaySafeLogger implements Logger {
+  private readonly context: ReplayContext;
+  private readonly innerLogger: Logger;
+
+  /**
+   * Creates a new ReplaySafeLogger.
+   *
+   * @param context - The replay context used to determine if replay is occurring.
+   * @param logger - The underlying logger to delegate to when not replaying.
+   */
+  constructor(context: ReplayContext, logger: Logger) {
+    if (!context) {
+      throw new Error("context is required");
+    }
+    if (!logger) {
+      throw new Error("logger is required");
+    }
+    this.context = context;
+    this.innerLogger = logger;
+  }
+
+  /**
+   * Logs an error message if not replaying.
+   * @param message - The error message to log.
+   * @param args - Additional arguments to include in the log.
+   */
+  error(message: string, ...args: unknown[]): void {
+    if (!this.context.isReplaying) {
+      this.innerLogger.error(message, ...args);
+    }
+  }
+
+  /**
+   * Logs a warning message if not replaying.
+   * @param message - The warning message to log.
+   * @param args - Additional arguments to include in the log.
+   */
+  warn(message: string, ...args: unknown[]): void {
+    if (!this.context.isReplaying) {
+      this.innerLogger.warn(message, ...args);
+    }
+  }
+
+  /**
+   * Logs an informational message if not replaying.
+   * @param message - The informational message to log.
+   * @param args - Additional arguments to include in the log.
+   */
+  info(message: string, ...args: unknown[]): void {
+    if (!this.context.isReplaying) {
+      this.innerLogger.info(message, ...args);
+    }
+  }
+
+  /**
+   * Logs a debug message if not replaying.
+   * @param message - The debug message to log.
+   * @param args - Additional arguments to include in the log.
+   */
+  debug(message: string, ...args: unknown[]): void {
+    if (!this.context.isReplaying) {
+      this.innerLogger.debug(message, ...args);
+    }
+  }
+}

packages/durabletask-js/src/utils/pb-helper.util.ts

@@ -306,11 +306,15 @@ export function newScheduleTaskAction(
   name: string,
   encodedInput?: string,
   tags?: Record<string, string>,
+  version?: string,
 ): pb.OrchestratorAction {
   const scheduleTaskAction = new pb.ScheduleTaskAction();
   scheduleTaskAction.setName(name);
   scheduleTaskAction.setInput(getStringValue(encodedInput));
   populateTagsMap(scheduleTaskAction.getTagsMap(), tags);
+  if (version) {
+    scheduleTaskAction.setVersion(getStringValue(version));
+  }
 
   const action = new pb.OrchestratorAction();
   action.setId(id);
@@ -331,12 +335,16 @@ export function newCreateSubOrchestrationAction(
   instanceId?: string | null,
   encodedInput?: string,
   tags?: Record<string, string>,
+  version?: string,
 ): pb.OrchestratorAction {
   const createSubOrchestrationAction = new pb.CreateSubOrchestrationAction();
   createSubOrchestrationAction.setName(name);
   createSubOrchestrationAction.setInstanceid(instanceId || "");
   createSubOrchestrationAction.setInput(getStringValue(encodedInput));
   populateTagsMap(createSubOrchestrationAction.getTagsMap(), tags);
+  if (version) {
+    createSubOrchestrationAction.setVersion(getStringValue(version));
+  }
 
   const action = new pb.OrchestratorAction();
   action.setId(id);