feat: Implement AsyncRetryHandler and refactor retry task handling

- Introduced RetryHandlerTask to support imperative retry control using AsyncRetryHandler.
- Refactored RuntimeOrchestrationContext to utilize createRetryTaskOrDefault for task creation.
- Enhanced retry logic to handle both RetryableTask and RetryHandlerTask.
- Added comprehensive tests for RetryHandlerTask, covering various scenarios including success, failure, and retry logic.
- Updated orchestration executor tests to validate new retry handler functionality.

Author: YunchuWang

packages/durabletask-js/src/task/retry-handler-task.ts

@@ -0,0 +1,83 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+import * as pb from "../proto/orchestrator_service_pb";
+import { RetryTaskBase } from "./retry-task-base";
+import { AsyncRetryHandler } from "./retry/retry-handler";
+import { createRetryContext } from "./retry/retry-context";
+import { TaskFailureDetails } from "./failure-details";
+
+/**
+ * A task that uses an AsyncRetryHandler for imperative retry control.
+ *
+ * @remarks
+ * Unlike RetryableTask which uses a declarative RetryPolicy, this task delegates
+ * all retry decisions to a user-provided handler function. The handler receives
+ * a RetryContext with failure details, attempt count, and elapsed time, and
+ * returns true to retry or false to stop.
+ *
+ * This mirrors the .NET SDK's InvokeWithCustomRetryHandler pattern, where the
+ * retry handler runs as orchestrator code (subject to replay).
+ */
+export class RetryHandlerTask<T> extends RetryTaskBase<T> {
+  private readonly _handler: AsyncRetryHandler;
+
+  /**
+   * Creates a new RetryHandlerTask instance.
+   *
+   * @param handler - The async retry handler for imperative retry decisions
+   * @param action - The orchestrator action associated with this task
+   * @param startTime - The time when the task was first scheduled
+   * @param taskType - The type of task (activity or sub-orchestration)
+   */
+  constructor(
+    handler: AsyncRetryHandler,
+    action: pb.OrchestratorAction,
+    startTime: Date,
+    taskType: "activity" | "subOrchestration",
+  ) {
+    super(action, startTime, taskType);
+    this._handler = handler;
+  }
+
+  /**
+   * Gets the async retry handler for this task.
+   */
+  get handler(): AsyncRetryHandler {
+    return this._handler;
+  }
+
+  /**
+   * Invokes the async retry handler to determine whether to retry.
+   *
+   * @param currentTime - The current orchestration time (for deterministic replay)
+   * @returns A Promise that resolves to true if the handler says to retry, false otherwise
+   */
+  async shouldRetry(currentTime: Date): Promise<boolean> {
+    if (!this.lastFailure) {
+      return false;
+    }
+
+    // Check for non-retriable failures (e.g., activity not found)
+    if (this.lastFailure.getIsnonretriable()) {
+      return false;
+    }
+
+    const failureDetails: TaskFailureDetails = {
+      errorType: this.lastFailure.getErrortype() || "Error",
+      message: this.lastFailure.getErrormessage() || "",
+      stackTrace: this.lastFailure.getStacktrace()?.getValue(),
+    };
+
+    const totalRetryTimeMs = currentTime.getTime() - this.startTime.getTime();
+
+    const retryContext = createRetryContext(
+      this.attemptCount,
+      failureDetails,
+      totalRetryTimeMs,
+      false, // isCancelled - not yet supported
+    );
+
+    return this._handler(retryContext);
+  }
+}

packages/durabletask-js/src/task/retry-task-base.ts

@@ -0,0 +1,139 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+import * as pb from "../proto/orchestrator_service_pb";
+import { TaskFailedError } from "./exception/task-failed-error";
+import { CompletableTask } from "./completable-task";
+
+/**
+ * Represents the type of retryable task - activity or sub-orchestration.
+ */
+export type RetryTaskType = "activity" | "subOrchestration";
+
+/**
+ * Abstract base class for tasks that support retry behavior.
+ *
+ * @remarks
+ * This class provides shared state and behavior for both policy-based retry
+ * (RetryableTask) and handler-based retry (RetryHandlerTask). It manages the
+ * action, attempt count, start time, task type, and failure tracking that are
+ * common to both retry strategies.
+ */
+export abstract class RetryTaskBase<T> extends CompletableTask<T> {
+  private readonly _action: pb.OrchestratorAction;
+  private readonly _startTime: Date;
+  private readonly _taskType: RetryTaskType;
+  private _attemptCount: number;
+  private _lastFailure: pb.TaskFailureDetails | undefined;
+
+  /**
+   * Creates a new RetryTaskBase instance.
+   *
+   * @param action - The orchestrator action associated with this task
+   * @param startTime - The time when the task was first scheduled
+   * @param taskType - The type of task (activity or sub-orchestration)
+   */
+  constructor(
+    action: pb.OrchestratorAction,
+    startTime: Date,
+    taskType: RetryTaskType,
+  ) {
+    super();
+    this._action = action;
+    this._startTime = startTime;
+    this._taskType = taskType;
+    this._attemptCount = 1;
+  }
+
+  /**
+   * Gets the orchestrator action associated with this task.
+   */
+  get action(): pb.OrchestratorAction {
+    return this._action;
+  }
+
+  /**
+   * Gets the current attempt count.
+   */
+  get attemptCount(): number {
+    return this._attemptCount;
+  }
+
+  /**
+   * Gets the time when the task was first scheduled.
+   */
+  get startTime(): Date {
+    return this._startTime;
+  }
+
+  /**
+   * Gets the type of task (activity or sub-orchestration).
+   */
+  get taskType(): RetryTaskType {
+    return this._taskType;
+  }
+
+  /**
+   * Gets the last failure details.
+   */
+  get lastFailure(): pb.TaskFailureDetails | undefined {
+    return this._lastFailure;
+  }
+
+  /**
+   * Increments the attempt count.
+   */
+  incrementAttemptCount(): void {
+    this._attemptCount++;
+  }
+
+  /**
+   * Records a failure for potential retry.
+   *
+   * @param message - The failure message
+   * @param details - The failure details from the protobuf
+   */
+  recordFailure(message: string, details?: pb.TaskFailureDetails): void {
+    details = details ?? new pb.TaskFailureDetails();
+    this._lastFailure = details;
+
+    // Store the exception for later if we exhaust retries
+    this._exception = new TaskFailedError(message, details);
+  }
+
+  /**
+   * Completes the task with the given result.
+   * Clears any previously recorded failure since the retry succeeded.
+   *
+   * @param result - The result of the task
+   */
+  override complete(result: T): void {
+    // Clear any previously recorded failure since the retry succeeded
+    this._exception = undefined;
+    this._lastFailure = undefined;
+
+    // Call the parent implementation
+    super.complete(result);
+  }
+
+  /**
+   * Marks the task as failed after all retry attempts are exhausted.
+   *
+   * @param message - The failure message
+   * @param details - Optional failure details
+   */
+  override fail(message: string, details?: pb.TaskFailureDetails): void {
+    if (this._isComplete) {
+      throw new Error("Task is already completed");
+    }
+
+    details = details ?? new pb.TaskFailureDetails();
+
+    this._exception = new TaskFailedError(message, details);
+    this._isComplete = true;
+
+    if (this._parent) {
+      this._parent.onChildCompleted(this);
+    }
+  }
+}