diff --git a/.github/workflows/dts-e2e-tests.yaml b/.github/workflows/dts-e2e-tests.yaml index ede600b6..67099aaa 100644 --- a/.github/workflows/dts-e2e-tests.yaml +++ b/.github/workflows/dts-e2e-tests.yaml @@ -4,6 +4,7 @@ name: ๐Ÿงช DTS Emulator E2E Tests # Tests are split across parallel jobs with separate task hubs for isolation. on: + workflow_dispatch: push: branches: - main @@ -24,7 +25,7 @@ jobs: - name: "entity" pattern: "test/e2e-azuremanaged/entity.spec.ts" - name: "orchestration" - pattern: "test/e2e-azuremanaged/orchestration.spec.ts" + pattern: "test/e2e-azuremanaged/orchestration.spec.ts test/e2e-azuremanaged/orchestration-id-reuse-policy.spec.ts" - name: "query-restart" pattern: "test/e2e-azuremanaged/query-apis.spec.ts test/e2e-azuremanaged/restart.spec.ts" - name: "retry-history-rewind" diff --git a/CHANGELOG.md b/CHANGELOG.md index 3eca2ca9..46e44627 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,10 +3,15 @@ ### New - Implement entity support in the in-memory testing backend ([#341](https://github.com/microsoft/durabletask-js/pull/341)) +- Add the top-level `StartOrchestrationOptions.dedupeStatuses` option, `ValidDedupeStatuses`, and + `OrchestrationAlreadyExistsError`, aligned with the .NET status-based duplicate rejection and + atomic replacement contract. The gRPC client defers omitted policies to its backend, while the + in-memory test client mirrors the .NET shim by treating omission as all statuses reusable. The + shared protocol does not support atomic no-op/`IGNORE`. +- Add the `CANCELED` member to the public `OrchestrationStatus` enum. ### Fixes - ## v0.4.0 (2026-07-31) ### Changes @@ -72,7 +77,7 @@ - fix: clear customStatus on continue-as-new in InMemoryOrchestrationBackend ([#155](https://github.com/microsoft/durabletask-js/pull/155)) - fix: propagate parent notification from composite tasks (WhenAllTask/WhenAnyTask) ([#150](https://github.com/microsoft/durabletask-js/pull/150)) - fix: use deterministic time in createTimer instead of Date.now() ([#146](https://github.com/microsoft/durabletask-js/pull/146)) -- Fix WhenAllTask constructor resetting _completedTasks counter ([#143](https://github.com/microsoft/durabletask-js/pull/143)) +- Fix WhenAllTask constructor resetting \_completedTasks counter ([#143](https://github.com/microsoft/durabletask-js/pull/143)) - Fix retry handler treating undefined/null/NaN/Infinity as retry signal ([#142](https://github.com/microsoft/durabletask-js/pull/142)) - Release v0.3.0 ([#147](https://github.com/microsoft/durabletask-js/pull/147)) diff --git a/README.md b/README.md index 683638b2..aeaf9350 100644 --- a/README.md +++ b/README.md @@ -27,9 +27,9 @@ const entityResponseBytes = await worker.processEntityBatchRequest(entityBatchRe The following npm packages are available for download. -| Name | Latest version | Description | -| - | - | - | -| Core SDK | [![npm version](https://img.shields.io/npm/v/@microsoft/durabletask-js)](https://www.npmjs.com/package/@microsoft/durabletask-js) | Core Durable Task SDK for JavaScript/TypeScript. | +| Name | Latest version | Description | +| ---------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| Core SDK | [![npm version](https://img.shields.io/npm/v/@microsoft/durabletask-js)](https://www.npmjs.com/package/@microsoft/durabletask-js) | Core Durable Task SDK for JavaScript/TypeScript. | | AzureManaged SDK | [![npm version](https://img.shields.io/npm/v/@microsoft/durabletask-js-azuremanaged)](https://www.npmjs.com/package/@microsoft/durabletask-js-azuremanaged) | Azure-managed [Durable Task Scheduler](https://learn.microsoft.com/azure/azure-functions/durable/durable-functions-task-scheduler) support for the Durable Task JavaScript SDK. | ## Prerequisites @@ -50,15 +50,8 @@ npm install @microsoft/durabletask-js @microsoft/durabletask-js-azuremanaged You can then use the following code to define a simple "Hello, cities" durable orchestration. ```typescript -import { - ActivityContext, - OrchestrationContext, - TOrchestrator, -} from "@microsoft/durabletask-js"; -import { - createAzureManagedClient, - createAzureManagedWorkerBuilder, -} from "@microsoft/durabletask-js-azuremanaged"; +import { ActivityContext, OrchestrationContext, TOrchestrator } from "@microsoft/durabletask-js"; +import { createAzureManagedClient, createAzureManagedWorkerBuilder } from "@microsoft/durabletask-js-azuremanaged"; // Define an activity function const sayHello = async (_: ActivityContext, name: string): Promise => { @@ -90,6 +83,32 @@ console.log(`Result: ${state?.serializedOutput}`); You can find more samples in the [examples/azure-managed](./examples/azure-managed) directory. +### Reusing orchestration instance IDs + +Set the top-level `dedupeStatuses` start option when an instance ID may be reused. The list +contains the existing runtime statuses that must continue to produce an +`OrchestrationAlreadyExistsError`; +instances in every other supported runtime status are atomically replaced: + +```typescript +import { OrchestrationStatus } from "@microsoft/durabletask-js"; + +await client.scheduleNewOrchestration(helloCities, undefined, { + instanceId: "daily-greeting", + dedupeStatuses: [OrchestrationStatus.RUNNING, OrchestrationStatus.PENDING], +}); +``` + +For `TaskHubGrpcClient`, omitting `dedupeStatuses` preserves the backend's default duplicate-ID +behavior; passing `[]` makes every supported runtime status replaceable. The in-memory +`TestOrchestrationClient` mirrors the .NET shim, where omission also makes all statuses reusable. +`ValidDedupeStatuses` exports the seven supported statuses. The transient `CONTINUED_AS_NEW` +status is not replaceable. A list containing `TERMINATED` must also contain `RUNNING`, `PENDING`, +and `SUSPENDED`, because replacing a running instance first terminates it. The production client +forwards this validation to the backend and maps its `INVALID_ARGUMENT` response to `TypeError`; +the in-memory client validates it directly. The current shared protocol does not define a +no-op/`IGNORE` action: a matching dedupe status is an error, while a non-matching status is replaced. + ## Supported patterns The following orchestration patterns are supported. @@ -125,10 +144,7 @@ An orchestration can wait for external events, such as a human approval, with op ```typescript import { whenAny } from "@microsoft/durabletask-js"; -const purchaseOrderWorkflow: TOrchestrator = async function* ( - ctx: OrchestrationContext, - order: Order, -): any { +const purchaseOrderWorkflow: TOrchestrator = async function* (ctx: OrchestrationContext, order: Order): any { // Orders under $1000 are auto-approved if (order.cost < 1000) { return "Auto-approved"; diff --git a/packages/azure-functions-durable/CHANGELOG.md b/packages/azure-functions-durable/CHANGELOG.md index 05fc47f7..880f1bd6 100644 --- a/packages/azure-functions-durable/CHANGELOG.md +++ b/packages/azure-functions-durable/CHANGELOG.md @@ -8,6 +8,8 @@ Interactive scenarios (external events, termination, suspend/resume) and entity batches are covered by driving the `@microsoft/durabletask-js` in-memory test stack with `wrapOrchestrator` / `wrapEntity`; see the README. +- Forward the top-level `dedupeStatuses` duplicate rejection and atomic replacement option through + `DurableFunctionsClient.startNew()`; the shared protocol does not support atomic no-op/`IGNORE`. ### Fixes diff --git a/packages/azure-functions-durable/README.md b/packages/azure-functions-durable/README.md index 19694078..d672a302 100644 --- a/packages/azure-functions-durable/README.md +++ b/packages/azure-functions-durable/README.md @@ -37,7 +37,9 @@ changed: and throws when the instance is missing. `showInput` suppresses only the top-level input, `showHistory` populates `history`, and `showHistoryOutput` toggles the per-entry input/result payloads; `history` entries are core `HistoryEvent`s (v3 types `history` as `Array`). - **`client.startNew()` supports the `version` option.** + **`client.startNew()` supports the `version` and top-level `dedupeStatuses` options.** Dedupe + statuses select duplicate errors; other supported statuses are atomically replaced. The current + shared protocol does not expose an atomic no-op/`IGNORE` action. - **Entity locking / critical sections moved to the core context.** v3's `context.df.lock(...)` / `context.df.isLocked()` and the `DurableLock` / `LockState` / `LockingRulesViolationError` exports are removed. Locks live on the core-native `context.entities` surface, which the classic diff --git a/packages/azure-functions-durable/src/client.ts b/packages/azure-functions-durable/src/client.ts index 2ce23102..37cdb53c 100644 --- a/packages/azure-functions-durable/src/client.ts +++ b/packages/azure-functions-durable/src/client.ts @@ -73,6 +73,11 @@ export interface StartNewOptions { instanceId?: string; /** Orchestration version to assign (forwarded to the core scheduler). */ version?: string; + /** + * Existing orchestration statuses that must produce a duplicate-ID error. + * The current shared protocol does not support an atomic no-op/IGNORE action. + */ + dedupeStatuses?: readonly OrchestrationStatus[]; } /** @@ -213,8 +218,12 @@ export class DurableFunctionsClient extends TaskHubGrpcClient { */ async startNew(orchestratorName: string, options?: StartNewOptions): Promise { const scheduleOptions = - options?.instanceId !== undefined || options?.version !== undefined - ? { instanceId: options?.instanceId, version: options?.version } + options?.instanceId !== undefined || options?.version !== undefined || options?.dedupeStatuses !== undefined + ? { + instanceId: options?.instanceId, + version: options?.version, + dedupeStatuses: options?.dedupeStatuses, + } : undefined; return this.scheduleNewOrchestration(orchestratorName, options?.input, scheduleOptions); } diff --git a/packages/azure-functions-durable/src/index.ts b/packages/azure-functions-durable/src/index.ts index 0b6fe23f..81862725 100644 --- a/packages/azure-functions-durable/src/index.ts +++ b/packages/azure-functions-durable/src/index.ts @@ -13,6 +13,7 @@ export { DurableFunctionsClientConfig, DurableFunctionsClientInput, DurableOrchestrationClient, + StartNewOptions, TaskHubOptions, getGrpcHostAddress, } from "./client"; @@ -22,6 +23,11 @@ export { createAzureFunctionsMetadataGenerator } from "./metadata"; export { DurableFunctionsWorker } from "./worker"; export { DurableBindingMetadata, addDurableGrpcMetadata } from "./durable-grpc"; export { RetryOptions } from "./retry-options"; +export { + OrchestrationAlreadyExistsError, + StartOrchestrationOptions, + ValidDedupeStatuses, +} from "@microsoft/durabletask-js"; // Re-exported core error so callers can `instanceof`-guard caught orchestration failures, matching // the classic durable-functions v3 top-level `TaskFailedError` export. (`DurableError` / // `AggregatedError` were never v3 top-level exports; the core engine surfaces `TaskFailedError` and diff --git a/packages/azure-functions-durable/src/orchestration-status.ts b/packages/azure-functions-durable/src/orchestration-status.ts index aa800ffb..0b325aa0 100644 --- a/packages/azure-functions-durable/src/orchestration-status.ts +++ b/packages/azure-functions-durable/src/orchestration-status.ts @@ -69,6 +69,8 @@ export function toOrchestrationRuntimeStatus(status: OrchestrationStatus): Orche return OrchestrationRuntimeStatus.ContinuedAsNew; case OrchestrationStatus.FAILED: return OrchestrationRuntimeStatus.Failed; + case OrchestrationStatus.CANCELED: + return OrchestrationRuntimeStatus.Canceled; case OrchestrationStatus.TERMINATED: return OrchestrationRuntimeStatus.Terminated; case OrchestrationStatus.PENDING: @@ -91,8 +93,9 @@ export function fromOrchestrationRuntimeStatus(status: OrchestrationRuntimeStatu return OrchestrationStatus.CONTINUED_AS_NEW; case OrchestrationRuntimeStatus.Failed: return OrchestrationStatus.FAILED; - case OrchestrationRuntimeStatus.Terminated: case OrchestrationRuntimeStatus.Canceled: + return OrchestrationStatus.CANCELED; + case OrchestrationRuntimeStatus.Terminated: return OrchestrationStatus.TERMINATED; case OrchestrationRuntimeStatus.Suspended: return OrchestrationStatus.SUSPENDED; diff --git a/packages/azure-functions-durable/test/unit/client.spec.ts b/packages/azure-functions-durable/test/unit/client.spec.ts index 22193e00..7bebfeda 100644 --- a/packages/azure-functions-durable/test/unit/client.spec.ts +++ b/packages/azure-functions-durable/test/unit/client.spec.ts @@ -384,6 +384,27 @@ describe("DurableFunctionsClient", () => { await client.stop(); } }); + + it("startNew forwards dedupe statuses to the core scheduler", async () => { + const client = new DurableFunctionsClient(CLIENT_CONFIG); + try { + const schedule = jest.spyOn(client, "scheduleNewOrchestration").mockResolvedValue("inst-9"); + const dedupeStatuses = [OrchestrationStatus.RUNNING]; + + await client.startNew("MyOrch", { + instanceId: "inst-9", + dedupeStatuses, + }); + + expect(schedule).toHaveBeenCalledWith("MyOrch", undefined, { + instanceId: "inst-9", + version: undefined, + dedupeStatuses, + }); + } finally { + await client.stop(); + } + }); }); describe("control-plane error mapping (v3 parity)", () => { @@ -403,7 +424,10 @@ describe("DurableFunctionsClient", () => { jest .spyOn(client, "terminateOrchestration") .mockRejectedValue( - grpcError(grpcStatus.NOT_FOUND, "5 NOT_FOUND: No instance with ID 'inst-1' was found. (Parameter 'instanceId')"), + grpcError( + grpcStatus.NOT_FOUND, + "5 NOT_FOUND: No instance with ID 'inst-1' was found. (Parameter 'instanceId')", + ), ); // terminate surfaces NOT_FOUND(5) for a genuinely missing instance (terminal instances get // FAILED_PRECONDITION instead), so the mapper consults getStatus: no state -> not-found message. @@ -543,7 +567,9 @@ describe("DurableFunctionsClient", () => { .spyOn(client, "resumeOrchestration") .mockRejectedValue(grpcError(grpcStatus.UNKNOWN, "2 UNKNOWN: Exception was thrown by handler")); jest.spyOn(client, "getOrchestrationState").mockResolvedValue(makeStateWith(OrchestrationStatus.RUNNING)); - await expect(client.resume("inst-1")).rejects.toThrow("Cannot resume orchestration instance in the Running state."); + await expect(client.resume("inst-1")).rejects.toThrow( + "Cannot resume orchestration instance in the Running state.", + ); } finally { await client.stop(); } diff --git a/packages/azure-functions-durable/test/unit/query-types.spec.ts b/packages/azure-functions-durable/test/unit/query-types.spec.ts index 9965d7f2..1ea1a562 100644 --- a/packages/azure-functions-durable/test/unit/query-types.spec.ts +++ b/packages/azure-functions-durable/test/unit/query-types.spec.ts @@ -6,6 +6,7 @@ import { EntityStateResponse } from "../../src/entity-state-response"; import { DurableOrchestrationStatus, OrchestrationRuntimeStatus, + fromOrchestrationRuntimeStatus, toDurableOrchestrationStatus, toOrchestrationRuntimeStatus, } from "../../src/orchestration-status"; @@ -25,6 +26,9 @@ describe("toOrchestrationRuntimeStatus", () => { expect(toOrchestrationRuntimeStatus(OrchestrationStatus.FAILED)).toBe( OrchestrationRuntimeStatus.Failed, ); + expect(toOrchestrationRuntimeStatus(OrchestrationStatus.CANCELED)).toBe( + OrchestrationRuntimeStatus.Canceled, + ); expect(toOrchestrationRuntimeStatus(OrchestrationStatus.TERMINATED)).toBe( OrchestrationRuntimeStatus.Terminated, ); @@ -37,6 +41,17 @@ describe("toOrchestrationRuntimeStatus", () => { }); }); +describe("fromOrchestrationRuntimeStatus", () => { + it("preserves the distinction between canceled and terminated", () => { + expect(fromOrchestrationRuntimeStatus(OrchestrationRuntimeStatus.Canceled)).toBe( + OrchestrationStatus.CANCELED, + ); + expect(fromOrchestrationRuntimeStatus(OrchestrationRuntimeStatus.Terminated)).toBe( + OrchestrationStatus.TERMINATED, + ); + }); +}); + describe("toDurableOrchestrationStatus", () => { it("maps a core OrchestrationState and deserializes JSON payloads", () => { const created = new Date("2026-01-01T00:00:00.000Z"); diff --git a/packages/durabletask-js-export-history/src/activities/export-instance-history-activity.ts b/packages/durabletask-js-export-history/src/activities/export-instance-history-activity.ts index 1c8f5fca..e715f70d 100644 --- a/packages/durabletask-js-export-history/src/activities/export-instance-history-activity.ts +++ b/packages/durabletask-js-export-history/src/activities/export-instance-history-activity.ts @@ -149,6 +149,7 @@ export function createExportInstanceHistoryActivity( OrchestrationStatus.COMPLETED, OrchestrationStatus.FAILED, OrchestrationStatus.TERMINATED, + OrchestrationStatus.CANCELED, ]; if (!terminalStatuses.includes(metadata.runtimeStatus)) { return { diff --git a/packages/durabletask-js-export-history/src/models/export-job-creation-options.ts b/packages/durabletask-js-export-history/src/models/export-job-creation-options.ts index c09df407..240ca4d8 100644 --- a/packages/durabletask-js-export-history/src/models/export-job-creation-options.ts +++ b/packages/durabletask-js-export-history/src/models/export-job-creation-options.ts @@ -85,10 +85,7 @@ export function createExportJobCreationOptions( ); } if (!options.completedTimeTo) { - throw new ExportJobClientValidationError( - "CompletedTimeTo is required for Batch export mode.", - "completedTimeTo", - ); + throw new ExportJobClientValidationError("CompletedTimeTo is required for Batch export mode.", "completedTimeTo"); } if (options.completedTimeTo <= options.completedTimeFrom) { throw new ExportJobClientValidationError( @@ -129,6 +126,7 @@ export function createExportJobCreationOptions( OrchestrationStatus.COMPLETED, OrchestrationStatus.FAILED, OrchestrationStatus.TERMINATED, + OrchestrationStatus.CANCELED, ]; if ( options.runtimeStatus && @@ -136,24 +134,19 @@ export function createExportJobCreationOptions( options.runtimeStatus.some((s) => !terminalStatuses.includes(s)) ) { throw new ExportJobClientValidationError( - "Export supports terminal orchestration statuses only. Valid statuses are: Completed, Failed, and Terminated.", + "Export supports terminal orchestration statuses only. Valid statuses are: Completed, Failed, Terminated, and Canceled.", "runtimeStatus", ); } // Default runtimeStatus to all terminal statuses if not provided const runtimeStatus = - options.runtimeStatus && options.runtimeStatus.length > 0 - ? options.runtimeStatus - : terminalStatuses; + options.runtimeStatus && options.runtimeStatus.length > 0 ? options.runtimeStatus : terminalStatuses; // Validate maxParallelExports range const validatedMaxParallelExports = options.maxParallelExports ?? DEFAULT_MAX_PARALLEL_EXPORTS; if (validatedMaxParallelExports <= 0) { - throw new ExportJobClientValidationError( - "MaxParallelExports must be greater than 0.", - "maxParallelExports", - ); + throw new ExportJobClientValidationError("MaxParallelExports must be greater than 0.", "maxParallelExports"); } return { diff --git a/packages/durabletask-js-export-history/test/models.spec.ts b/packages/durabletask-js-export-history/test/models.spec.ts index a78b5ce9..e3ba981b 100644 --- a/packages/durabletask-js-export-history/test/models.spec.ts +++ b/packages/durabletask-js-export-history/test/models.spec.ts @@ -81,6 +81,7 @@ describe("Models", () => { OrchestrationStatus.COMPLETED, OrchestrationStatus.FAILED, OrchestrationStatus.TERMINATED, + OrchestrationStatus.CANCELED, ]); }); diff --git a/packages/durabletask-js/src/client/client.ts b/packages/durabletask-js/src/client/client.ts index af90fc7c..9f0c3682 100644 --- a/packages/durabletask-js/src/client/client.ts +++ b/packages/durabletask-js/src/client/client.ts @@ -27,6 +27,8 @@ import { HistoryEvent } from "../orchestration/history-event"; import { convertProtoHistoryEvent } from "../utils/history-event-converter"; import { Logger, ConsoleLogger } from "../types/logger.type"; import { StartOrchestrationOptions } from "../task/options"; +import { toProtobufOrchestrationIdReusePolicy } from "../orchestration/orchestration-id-reuse-policy"; +import { OrchestrationAlreadyExistsError } from "../orchestration/exception/orchestration-already-exists-error"; import { mapToRecord } from "../utils/tags.util"; import { populateTagsMap } from "../utils/pb-helper.util"; import { EntityInstanceId } from "../entities/entity-instance-id"; @@ -174,7 +176,7 @@ export class TaskHubGrpcClient { * * @param {TOrchestrator | string} orchestrator - The orchestrator or the name of the orchestrator to be scheduled. * @param {TInput} input - Optional input for the orchestrator. - * @param {StartOrchestrationOptions} options - Options for instance ID, start time, and tags. + * @param {StartOrchestrationOptions} options - Options for instance ID, start time, tags, version, and ID reuse. * @return {Promise} A Promise resolving to the unique ID of the scheduled orchestrator instance. */ async scheduleNewOrchestration( @@ -211,6 +213,10 @@ export class TaskHubGrpcClient { typeof instanceIdOrOptions === "string" || instanceIdOrOptions === undefined ? undefined : instanceIdOrOptions.version; + const dedupeStatuses = + typeof instanceIdOrOptions === "string" || instanceIdOrOptions === undefined + ? undefined + : instanceIdOrOptions.dedupeStatuses; // Use provided version, or fall back to client's default version const effectiveVersion = version ?? this._defaultVersion; @@ -234,6 +240,10 @@ export class TaskHubGrpcClient { req.setVersion(v); } + if (dedupeStatuses !== undefined) { + req.setOrchestrationidreusepolicy(toProtobufOrchestrationIdReusePolicy(dedupeStatuses)); + } + populateTagsMap(req.getTagsMap(), tags); // Create a tracing span for the new orchestration (if OTEL is available) @@ -254,6 +264,17 @@ export class TaskHubGrpcClient { return res.getInstanceid(); } catch (e: unknown) { setSpanError(span, e); + if (e instanceof Error && "code" in e) { + const grpcError = e as grpc.ServiceError; + const message = grpcError.details || grpcError.message; + if (grpcError.code === grpc.status.ALREADY_EXISTS) { + throw new OrchestrationAlreadyExistsError(message, { cause: e }); + } + if (grpcError.code === grpc.status.INVALID_ARGUMENT) { + throw new TypeError(message, { cause: e }); + } + // The JS SDK has no named cancellation error contract, so preserve CANCELLED as a ServiceError. + } throw e; } finally { endSpan(span); @@ -575,7 +596,11 @@ export class TaskHubGrpcClient { throw new Error(`An orchestration with the instanceId '${instanceId}' was not found.`, { cause: e }); } if (grpcError.code === grpc.status.FAILED_PRECONDITION) { - throw new Error(grpcError.details || `Cannot rewind orchestration '${instanceId}': it is in a state that does not allow rewinding.`, { cause: e }); + throw new Error( + grpcError.details || + `Cannot rewind orchestration '${instanceId}': it is in a state that does not allow rewinding.`, + { cause: e }, + ); } if (grpcError.code === grpc.status.UNIMPLEMENTED) { throw new Error(grpcError.details || `The rewind operation is not supported by the backend.`, { cause: e }); @@ -702,7 +727,12 @@ export class TaskHubGrpcClient { req.setRecursive(options?.recursive ?? false); const timeout = purgeInstanceCriteria.getTimeout(); - ClientLogs.purgingInstances(this._logger, createdTimeFrom, createdTimeTo, runtimeStatusList.map(String).join(", ")); + ClientLogs.purgingInstances( + this._logger, + createdTimeFrom, + createdTimeTo, + runtimeStatusList.map(String).join(", "), + ); const callPromise = callWithMetadata( this._stub.purgeInstances.bind(this._stub), @@ -817,7 +847,10 @@ export class TaskHubGrpcClient { const states: OrchestrationState[] = []; const orchestrationStateList = response.getOrchestrationstateList(); for (const state of orchestrationStateList) { - const orchestrationState = this._createOrchestrationStateFromProto(state, filter?.fetchInputsAndOutputs ?? false); + const orchestrationState = this._createOrchestrationStateFromProto( + state, + filter?.fetchInputsAndOutputs ?? false, + ); if (orchestrationState) { states.push(orchestrationState); } @@ -970,7 +1003,11 @@ export class TaskHubGrpcClient { } else if (err.code === grpc.status.CANCELLED) { reject(new Error(`The getOrchestrationHistory operation was canceled.`)); } else if (err.code === grpc.status.INTERNAL) { - reject(new Error(`An error occurred while retrieving the history for orchestration with instanceId '${instanceId}'.`)); + reject( + new Error( + `An error occurred while retrieving the history for orchestration with instanceId '${instanceId}'.`, + ), + ); } else { reject(err); } @@ -1233,10 +1270,20 @@ export class TaskHubGrpcClient { return createEntityMetadata(entityId, lastModifiedTime, backlogQueueSize, lockedBy, state); } catch { // Return metadata without state if parsing fails - return createEntityMetadataWithoutState(entityId, lastModifiedTime, backlogQueueSize, lockedBy) as EntityMetadata; + return createEntityMetadataWithoutState( + entityId, + lastModifiedTime, + backlogQueueSize, + lockedBy, + ) as EntityMetadata; } } else { - return createEntityMetadataWithoutState(entityId, lastModifiedTime, backlogQueueSize, lockedBy) as EntityMetadata; + return createEntityMetadataWithoutState( + entityId, + lastModifiedTime, + backlogQueueSize, + lockedBy, + ) as EntityMetadata; } } @@ -1299,4 +1346,4 @@ export class TaskHubGrpcClient { tags, ); } -} \ No newline at end of file +} diff --git a/packages/durabletask-js/src/index.ts b/packages/durabletask-js/src/index.ts index 1bb5cd3d..bdf3e6ee 100644 --- a/packages/durabletask-js/src/index.ts +++ b/packages/durabletask-js/src/index.ts @@ -28,6 +28,8 @@ export { TERMINATE_OPTIONS_SYMBOL, } from "./orchestration/orchestration-terminate-options"; export { OrchestrationStatus } from "./orchestration/enum/orchestration-status.enum"; +export { ValidDedupeStatuses } from "./orchestration/orchestration-id-reuse-policy"; +export { OrchestrationAlreadyExistsError } from "./orchestration/exception/orchestration-already-exists-error"; export { OrchestrationState } from "./orchestration/orchestration-state"; // Query types diff --git a/packages/durabletask-js/src/orchestration/enum/orchestration-status.enum.ts b/packages/durabletask-js/src/orchestration/enum/orchestration-status.enum.ts index 64aff915..f79dc997 100644 --- a/packages/durabletask-js/src/orchestration/enum/orchestration-status.enum.ts +++ b/packages/durabletask-js/src/orchestration/enum/orchestration-status.enum.ts @@ -7,7 +7,6 @@ import * as pb from "../../proto/orchestrator_service_pb"; const protoToClient = new Map(); const clientToProto = new Map(); - export function fromProtobuf(val: pb.OrchestrationStatus): OrchestrationStatus { const result = protoToClient.get(val); if (result === undefined) { @@ -28,6 +27,7 @@ export enum OrchestrationStatus { RUNNING = pb.OrchestrationStatus.ORCHESTRATION_STATUS_RUNNING, COMPLETED = pb.OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED, FAILED = pb.OrchestrationStatus.ORCHESTRATION_STATUS_FAILED, + CANCELED = pb.OrchestrationStatus.ORCHESTRATION_STATUS_CANCELED, TERMINATED = pb.OrchestrationStatus.ORCHESTRATION_STATUS_TERMINATED, CONTINUED_AS_NEW = pb.OrchestrationStatus.ORCHESTRATION_STATUS_CONTINUED_AS_NEW, PENDING = pb.OrchestrationStatus.ORCHESTRATION_STATUS_PENDING, @@ -46,7 +46,7 @@ for (const [name, value] of Object.entries(OrchestrationStatus)) { if (protoValue !== numValue) { throw new Error( `Enum drift detected: OrchestrationStatus.${name} (${numValue}) does not match ` + - `pb.OrchestrationStatus.${expectedProtoKey} (${protoValue}).`, + `pb.OrchestrationStatus.${expectedProtoKey} (${protoValue}).`, ); } protoToClient.set(numValue as pb.OrchestrationStatus, numValue as OrchestrationStatus); diff --git a/packages/durabletask-js/src/orchestration/exception/orchestration-already-exists-error.ts b/packages/durabletask-js/src/orchestration/exception/orchestration-already-exists-error.ts new file mode 100644 index 00000000..c50f22fc --- /dev/null +++ b/packages/durabletask-js/src/orchestration/exception/orchestration-already-exists-error.ts @@ -0,0 +1,10 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +/** Thrown when an orchestration ID reuse request matches an existing dedupe status. */ +export class OrchestrationAlreadyExistsError extends Error { + constructor(message: string, options?: ErrorOptions) { + super(message, options); + this.name = "OrchestrationAlreadyExistsError"; + } +} diff --git a/packages/durabletask-js/src/orchestration/orchestration-id-reuse-policy.ts b/packages/durabletask-js/src/orchestration/orchestration-id-reuse-policy.ts new file mode 100644 index 00000000..25b39d0b --- /dev/null +++ b/packages/durabletask-js/src/orchestration/orchestration-id-reuse-policy.ts @@ -0,0 +1,55 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import * as pb from "../proto/orchestrator_service_pb"; +import { OrchestrationStatus, toProtobuf } from "./enum/orchestration-status.enum"; + +/** Runtime statuses supported by orchestration ID deduplication, matching the .NET SDK. */ +export const ValidDedupeStatuses: readonly OrchestrationStatus[] = Object.freeze([ + OrchestrationStatus.COMPLETED, + OrchestrationStatus.FAILED, + OrchestrationStatus.TERMINATED, + OrchestrationStatus.CANCELED, + OrchestrationStatus.PENDING, + OrchestrationStatus.RUNNING, + OrchestrationStatus.SUSPENDED, +]); + +/** @hidden Validates orchestration ID deduplication options. */ +export function validateDedupeStatuses(dedupeStatuses: readonly OrchestrationStatus[]): void { + for (const status of dedupeStatuses) { + if (!ValidDedupeStatuses.includes(status)) { + throw new TypeError(`Invalid orchestration runtime status: '${status}' for deduplication.`); + } + } +} + +/** @hidden Validates deduplication options for clients that terminate reusable running instances themselves. */ +export function validateDedupeStatusesForReplacement(dedupeStatuses: readonly OrchestrationStatus[]): void { + validateDedupeStatuses(dedupeStatuses); + const dedupeStatusSet = new Set(dedupeStatuses); + if ( + dedupeStatusSet.has(OrchestrationStatus.TERMINATED) && + [OrchestrationStatus.RUNNING, OrchestrationStatus.PENDING, OrchestrationStatus.SUSPENDED].some( + (status) => !dedupeStatusSet.has(status), + ) + ) { + throw new TypeError( + "Invalid dedupe statuses: cannot include 'Terminated' while also allowing reuse of running instances, " + + "because the running instance would be terminated and then immediately conflict with the dedupe check.", + ); + } +} + +/** @hidden Converts the public deduplication policy to the wire-level replacement policy. */ +export function toProtobufOrchestrationIdReusePolicy( + dedupeStatuses: readonly OrchestrationStatus[], +): pb.OrchestrationIdReusePolicy { + validateDedupeStatuses(dedupeStatuses); + const dedupeStatusSet = new Set(dedupeStatuses.map(toProtobuf)); + const replaceableStatuses = ValidDedupeStatuses.map(toProtobuf).filter((status) => !dedupeStatusSet.has(status)); + + const result = new pb.OrchestrationIdReusePolicy(); + result.setReplaceablestatusList(replaceableStatuses); + return result; +} diff --git a/packages/durabletask-js/src/task/options/task-options.ts b/packages/durabletask-js/src/task/options/task-options.ts index 765baa8a..e876e8f2 100644 --- a/packages/durabletask-js/src/task/options/task-options.ts +++ b/packages/durabletask-js/src/task/options/task-options.ts @@ -3,6 +3,7 @@ import { RetryPolicy } from "../retry/retry-policy"; import { AsyncRetryHandler, RetryHandler } from "../retry/retry-handler"; +import { OrchestrationStatus } from "../../orchestration/enum/orchestration-status.enum"; /** * Union type representing the available retry strategies for a task. @@ -74,6 +75,14 @@ export interface StartOrchestrationOptions { * via the OrchestrationContext.version property. */ version?: string; + /** + * Existing orchestration statuses that must produce a duplicate-ID error. + * + * An empty list makes every supported status replaceable. The gRPC client preserves + * the backend's default when this property is omitted; the in-memory test client + * mirrors the .NET shim by treating omission as all statuses reusable. + */ + dedupeStatuses?: readonly OrchestrationStatus[]; } /** diff --git a/packages/durabletask-js/src/testing/in-memory-backend.ts b/packages/durabletask-js/src/testing/in-memory-backend.ts index a05cde63..a0d9a498 100644 --- a/packages/durabletask-js/src/testing/in-memory-backend.ts +++ b/packages/durabletask-js/src/testing/in-memory-backend.ts @@ -7,6 +7,8 @@ import { OrchestrationStatus as ClientOrchestrationStatus } from "../orchestrati import { ParentOrchestrationInstance } from "../types/parent-orchestration-instance.type"; import { StringValue } from "google-protobuf/google/protobuf/wrappers_pb"; import { randomUUID } from "crypto"; +import { validateDedupeStatusesForReplacement } from "../orchestration/orchestration-id-reuse-policy"; +import { OrchestrationAlreadyExistsError } from "../orchestration/exception/orchestration-already-exists-error"; /** Mints a fresh per-execution ID (DTFx `Guid.ToString("N")` idiom: 32 hex chars, no dashes). */ function newExecutionId(): string { @@ -37,6 +39,7 @@ export interface OrchestrationInstance { */ export interface ActivityWorkItem { instanceId: string; + executionId: string; name: string; taskId: number; input?: string; @@ -98,16 +101,23 @@ interface StateWaiter { resolve: (instance: OrchestrationInstance | undefined) => void; reject: (error: Error) => void; predicate: (instance: OrchestrationInstance) => boolean; + subOrchestrationWatcher?: SubOrchestrationWatcher; +} + +interface SubOrchestrationWatcher { + parentInstanceId: string; + parentExecutionId: string; + taskId: number; } /** * In-memory backend for durable orchestrations suitable for testing. - * + * * This backend stores all orchestration state in memory and processes * work items synchronously within the same process. It is designed for * unit testing and integration testing scenarios where a sidecar process * or external storage is not desired. - * + * * Thread-safety: All state mutations are performed synchronously via * the event loop. The backend uses a simple work queue pattern to ensure * that orchestration and activity processing happens in a predictable order. @@ -150,7 +160,7 @@ export class InMemoryOrchestrationBackend { parentInstance?: ParentOrchestrationInstance, ): string { if (this.instances.has(instanceId)) { - throw new Error(`Orchestration instance '${instanceId}' already exists`); + throw new OrchestrationAlreadyExistsError(`An orchestration with instance ID '${instanceId}' already exists`); } const now = new Date(); @@ -186,6 +196,70 @@ export class InMemoryOrchestrationBackend { return instanceId; } + /** + * Creates a client-scheduled orchestration using the same status-based reuse semantics as the .NET test host. + */ + async createOrchestrationInstance( + instanceId: string, + name: string, + input?: string, + scheduledStartTime?: Date, + dedupeStatuses?: readonly ClientOrchestrationStatus[], + ): Promise { + if (dedupeStatuses !== undefined) { + validateDedupeStatusesForReplacement(dedupeStatuses); + } + + const existingInstance = this.instances.get(instanceId); + if (existingInstance) { + const existingStatus = this.toClientStatus(existingInstance.status); + if (dedupeStatuses?.includes(existingStatus) === true) { + throw this.newAlreadyExistsError(instanceId, existingStatus); + } + + const existingExecutionId = existingInstance.executionId; + if (this.isRunningStatus(existingInstance.status)) { + const dedupeDescription = + dedupeStatuses === undefined + ? "undefined (all statuses reusable)" + : dedupeStatuses.length === 0 + ? "[] (all statuses reusable)" + : `[${dedupeStatuses.join(", ")}]`; + const terminationReason = + `A new instance creation request has been issued for instance ${instanceId} which currently has status ` + + `${this.formatStatus(existingStatus)}. Since the dedupe statuses of the creation request, ` + + `${dedupeDescription}, do not contain the orchestration's status, the orchestration has been terminated ` + + "and a new instance with the same instance ID will be created."; + + const encodedTerminationReason = JSON.stringify(terminationReason); + this.terminate(instanceId, encodedTerminationReason); + this.completeOrchestration(instanceId, existingInstance.completionToken, [ + pbh.newCompleteOrchestrationAction( + -1, + pb.OrchestrationStatus.ORCHESTRATION_STATUS_TERMINATED, + encodedTerminationReason, + ), + ]); + await this.waitForState( + instanceId, + (instance) => instance.executionId === existingExecutionId && this.isTerminalStatus(instance.status), + 0, + ); + } + + const currentInstance = this.instances.get(instanceId); + if (currentInstance?.executionId === existingExecutionId) { + const currentStatus = this.toClientStatus(currentInstance.status); + if (dedupeStatuses?.includes(currentStatus) === true) { + throw this.newAlreadyExistsError(instanceId, currentStatus); + } + this.removeInstanceForReplacement(instanceId); + } + } + + return this.createInstance(instanceId, name, input, scheduledStartTime); + } + /** * Gets an orchestration instance by ID. */ @@ -362,7 +436,7 @@ export class InMemoryOrchestrationBackend { const instanceId = this.orchestrationQueue.shift()!; this.orchestrationQueueSet.delete(instanceId); const instance = this.instances.get(instanceId); - + if (instance && instance.pendingEvents.length > 0) { return instance; } @@ -623,9 +697,7 @@ export class InMemoryOrchestrationBackend { // Continue-as-new resets status to PENDING and rewind resets it to RUNNING, so neither is // terminal here and neither gets a bookend. if (this.isTerminalStatus(instance.status)) { - instance.history.push( - pbh.newExecutionCompletedEvent(instance.status, instance.output, instance.failureDetails), - ); + instance.history.push(pbh.newExecutionCompletedEvent(instance.status, instance.output, instance.failureDetails)); } // Update completion token for next execution @@ -638,16 +710,14 @@ export class InMemoryOrchestrationBackend { /** * Completes an activity execution. */ - completeActivity( - instanceId: string, - taskId: number, - result?: string, - error?: Error, - ): void { + completeActivity(instanceId: string, executionId: string, taskId: number, result?: string, error?: Error): void { const instance = this.instances.get(instanceId); if (!instance) { return; // Instance may have been purged } + if (instance.executionId !== executionId) { + return; // Completion belongs to a replaced or continued-as-new execution + } let event: pb.HistoryEvent; if (error) { @@ -668,6 +738,15 @@ export class InMemoryOrchestrationBackend { instanceId: string, predicate: (instance: OrchestrationInstance) => boolean, timeoutMs: number = 30000, + ): Promise { + return this.waitForStateInternal(instanceId, predicate, timeoutMs); + } + + private async waitForStateInternal( + instanceId: string, + predicate: (instance: OrchestrationInstance) => boolean, + timeoutMs: number, + subOrchestrationWatcher?: SubOrchestrationWatcher, ): Promise { const instance = this.instances.get(instanceId); if (instance && predicate(instance)) { @@ -676,7 +755,7 @@ export class InMemoryOrchestrationBackend { return new Promise((resolve, reject) => { // When timeoutMs is 0, no timeout is applied โ€” the waiter will only be - // resolved by a matching state change or rejected by reset(). + // resolved by a matching state change or rejected by lifecycle cleanup. // `waiter` is declared before the timer so the timeout callback can find it by // object identity; the waiter reads `timer` only when invoked, by which point // it has been assigned. @@ -698,6 +777,7 @@ export class InMemoryOrchestrationBackend { reject(error); }, predicate, + subOrchestrationWatcher, }; if (timeoutMs > 0) { @@ -777,6 +857,8 @@ export class InMemoryOrchestrationBackend { return ClientOrchestrationStatus.COMPLETED; case pb.OrchestrationStatus.ORCHESTRATION_STATUS_FAILED: return ClientOrchestrationStatus.FAILED; + case pb.OrchestrationStatus.ORCHESTRATION_STATUS_CANCELED: + return ClientOrchestrationStatus.CANCELED; case pb.OrchestrationStatus.ORCHESTRATION_STATUS_TERMINATED: return ClientOrchestrationStatus.TERMINATED; case pb.OrchestrationStatus.ORCHESTRATION_STATUS_SUSPENDED: @@ -796,11 +878,121 @@ export class InMemoryOrchestrationBackend { } } + private removeInstanceForReplacement(instanceId: string): void { + const instance = this.instances.get(instanceId); + if (!instance) { + return; + } + + this.cancelInstanceTimers(instanceId); + this.cancelSubOrchestrationWatchers( + instanceId, + instance.executionId, + new Error(`Parent orchestration instance '${instanceId}' was replaced by a new execution`), + ); + this.rejectStateWaitersForReplacement( + instanceId, + new Error(`Orchestration instance '${instanceId}' was replaced by a new execution`), + ); + this.instances.delete(instanceId); + this.orchestrationQueueSet.delete(instanceId); + + const orchestrationQueueIndex = this.orchestrationQueue.indexOf(instanceId); + if (orchestrationQueueIndex >= 0) { + this.orchestrationQueue.splice(orchestrationQueueIndex, 1); + } + + for (let i = this.activityQueue.length - 1; i >= 0; i--) { + if (this.activityQueue[i].instanceId === instanceId) { + this.activityQueue.splice(i, 1); + } + } + } + + private rejectStateWaitersForReplacement(instanceId: string, error: Error): void { + const waiters = this.stateWaiters.get(instanceId); + if (!waiters) { + return; + } + + this.stateWaiters.delete(instanceId); + for (const waiter of waiters) { + if (waiter.subOrchestrationWatcher) { + this.failSubOrchestrationWatcher( + waiter.subOrchestrationWatcher, + new Error(`Sub-orchestration instance '${instanceId}' was replaced by a new execution`), + ); + } + waiter.reject(error); + } + } + + private cancelSubOrchestrationWatchers(parentInstanceId: string, parentExecutionId: string, error: Error): void { + for (const [instanceId, waiters] of this.stateWaiters) { + const cancelledWaiters = waiters.filter( + (waiter) => + waiter.subOrchestrationWatcher?.parentInstanceId === parentInstanceId && + waiter.subOrchestrationWatcher.parentExecutionId === parentExecutionId, + ); + if (cancelledWaiters.length === 0) { + continue; + } + + const remainingWaiters = waiters.filter((waiter) => !cancelledWaiters.includes(waiter)); + if (remainingWaiters.length === 0) { + this.stateWaiters.delete(instanceId); + } else { + this.stateWaiters.set(instanceId, remainingWaiters); + } + for (const waiter of cancelledWaiters) { + waiter.reject(error); + } + } + } + + private failSubOrchestrationWatcher(watcher: SubOrchestrationWatcher, error: Error): void { + const parentInstance = this.instances.get(watcher.parentInstanceId); + if ( + !parentInstance || + parentInstance.executionId !== watcher.parentExecutionId || + this.isTerminalStatus(parentInstance.status) + ) { + return; + } + + parentInstance.pendingEvents.push(pbh.newSubOrchestrationFailedEvent(watcher.taskId, error)); + parentInstance.lastUpdatedAt = new Date(); + this.enqueueOrchestration(watcher.parentInstanceId); + } + + private newAlreadyExistsError( + instanceId: string, + status: ClientOrchestrationStatus, + ): OrchestrationAlreadyExistsError { + return new OrchestrationAlreadyExistsError( + `An orchestration with instance ID '${instanceId}' and status '${this.formatStatus(status)}' already exists`, + ); + } + + private formatStatus(status: ClientOrchestrationStatus): string { + const name = ClientOrchestrationStatus[status].toLowerCase(); + return name.charAt(0).toUpperCase() + name.slice(1); + } + + private isRunningStatus(status: pb.OrchestrationStatus): boolean { + return ( + status === pb.OrchestrationStatus.ORCHESTRATION_STATUS_RUNNING || + status === pb.OrchestrationStatus.ORCHESTRATION_STATUS_PENDING || + status === pb.OrchestrationStatus.ORCHESTRATION_STATUS_SUSPENDED + ); + } + private isTerminalStatus(status: pb.OrchestrationStatus): boolean { return ( status === pb.OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED || status === pb.OrchestrationStatus.ORCHESTRATION_STATUS_FAILED || - status === pb.OrchestrationStatus.ORCHESTRATION_STATUS_TERMINATED + status === pb.OrchestrationStatus.ORCHESTRATION_STATUS_TERMINATED || + status === pb.OrchestrationStatus.ORCHESTRATION_STATUS_CANCELED ); } @@ -886,7 +1078,13 @@ export class InMemoryOrchestrationBackend { // because it sets currentUtcDateTime, and ExecutionStarted must come before // carryover events because it initializes the orchestrator generator. const orchestratorStarted = pbh.newOrchestratorStartedEvent(new Date()); - const executionStarted = pbh.newExecutionStartedEvent(instance.name, instance.instanceId, newInput, undefined, instance.executionId); + const executionStarted = pbh.newExecutionStartedEvent( + instance.name, + instance.instanceId, + newInput, + undefined, + instance.executionId, + ); instance.pendingEvents = [orchestratorStarted, executionStarted, ...carryoverEvents]; this.enqueueOrchestration(instance.instanceId); @@ -911,6 +1109,7 @@ export class InMemoryOrchestrationBackend { // Queue activity for execution this.activityQueue.push({ instanceId: instance.instanceId, + executionId: instance.executionId, name: taskName, taskId, input, @@ -935,12 +1134,17 @@ export class InMemoryOrchestrationBackend { // Schedule timer firing const now = new Date(); const delay = Math.max(0, fireAt.getTime() - now.getTime()); + const executionId = instance.executionId; const timerHandle = setTimeout(() => { this.pendingTimers.delete(timerHandle); this.removeInstanceTimer(instance.instanceId, timerHandle); const currentInstance = this.instances.get(instance.instanceId); - if (currentInstance && !this.isTerminalStatus(currentInstance.status)) { + if ( + currentInstance && + currentInstance.executionId === executionId && + !this.isTerminalStatus(currentInstance.status) + ) { const timerFiredEvent = pbh.newTimerFiredEvent(timerId, fireAt); currentInstance.pendingEvents.push(timerFiredEvent); currentInstance.lastUpdatedAt = new Date(); @@ -976,7 +1180,7 @@ export class InMemoryOrchestrationBackend { }); // Watch for sub-orchestration completion - this.watchSubOrchestration(instance.instanceId, subInstanceId, taskId); + this.watchSubOrchestration(instance.instanceId, instance.executionId, subInstanceId, taskId); } catch (error: unknown) { // Sub-orchestration creation failed const err = error instanceof Error ? error : new Error(String(error)); @@ -986,19 +1190,25 @@ export class InMemoryOrchestrationBackend { } } - private watchSubOrchestration(parentInstanceId: string, subInstanceId: string, taskId: number): void { + private watchSubOrchestration( + parentInstanceId: string, + parentExecutionId: string, + subInstanceId: string, + taskId: number, + ): void { // Use the stateWaiters mechanism instead of polling to avoid infinite loops // and unnecessary resource consumption - this.waitForState( + this.waitForStateInternal( subInstanceId, (inst) => this.isTerminalStatus(inst.status), 0, // No timeout โ€” sub-orchestration will eventually complete, fail, or be terminated + { parentInstanceId, parentExecutionId, taskId }, ) .then((subInstance) => { const parentInstance = this.instances.get(parentInstanceId); // If parent or sub no longer exists, nothing to do - if (!subInstance || !parentInstance) { + if (!subInstance || !parentInstance || parentInstance.executionId !== parentExecutionId) { return; } @@ -1020,15 +1230,11 @@ export class InMemoryOrchestrationBackend { this.enqueueOrchestration(parentInstanceId); }) .catch(() => { - // Reset โ€” sub-orchestration watcher cancelled, nothing to do + // The watcher was cancelled by reset or parent/child replacement. }); } - private prepareRewind( - instance: OrchestrationInstance, - reason?: string, - snapshot?: RewindSnapshot, - ): void { + private prepareRewind(instance: OrchestrationInstance, reason?: string, snapshot?: RewindSnapshot): void { // Reset instance state so it can be re-processed. instance.status = pb.OrchestrationStatus.ORCHESTRATION_STATUS_RUNNING; instance.output = undefined; @@ -1143,7 +1349,7 @@ export class InMemoryOrchestrationBackend { this.prepareRewind(subInstance, reason, snapshot); } } - this.watchSubOrchestration(instance.instanceId, subInstanceId, taskId); + this.watchSubOrchestration(instance.instanceId, instance.executionId, subInstanceId, taskId); } // Re-enqueue so the orchestration replays with the clean history. The executionRewound @@ -1488,10 +1694,7 @@ export class InMemoryOrchestrationBackend { for (const event of history) { const lockRequested = event.getEntitylockrequested(); if (lockRequested) { - remainingLockSets.set( - lockRequested.getCriticalsectionid(), - new Set(lockRequested.getLocksetList()), - ); + remainingLockSets.set(lockRequested.getCriticalsectionid(), new Set(lockRequested.getLocksetList())); continue; } @@ -1515,10 +1718,7 @@ export class InMemoryOrchestrationBackend { return remainingLockSets.size > 0; } - private validateRewindLockState( - instance: OrchestrationInstance, - snapshot: RewindSnapshot, - ): void { + private validateRewindLockState(instance: OrchestrationInstance, snapshot: RewindSnapshot): void { if (snapshot.has(instance.instanceId)) { return; } @@ -1528,9 +1728,7 @@ export class InMemoryOrchestrationBackend { }); if (this.hasUnreleasedEntityLock(instance.history)) { - throw new Error( - `Cannot rewind an orchestration with an unreleased entity lock: '${instance.instanceId}'`, - ); + throw new Error(`Cannot rewind an orchestration with an unreleased entity lock: '${instance.instanceId}'`); } const completedSubOrchestrationTaskIds = new Set(); diff --git a/packages/durabletask-js/src/testing/test-client.ts b/packages/durabletask-js/src/testing/test-client.ts index 517fbca2..682a0480 100644 --- a/packages/durabletask-js/src/testing/test-client.ts +++ b/packages/durabletask-js/src/testing/test-client.ts @@ -11,10 +11,11 @@ import { EntityInstanceId } from "../entities/entity-instance-id"; import { EntityMetadata } from "../entities/entity-metadata"; import { InMemoryOrchestrationBackend, OrchestrationInstance } from "./in-memory-backend"; import * as pb from "../proto/orchestrator_service_pb"; +import { StartOrchestrationOptions } from "../task/options"; /** * Client for scheduling and managing orchestrations in the in-memory backend. - * + * * This client provides a similar API to TaskHubGrpcClient but operates * entirely in-memory for testing purposes. */ @@ -23,18 +24,52 @@ export class TestOrchestrationClient { /** * Schedules a new orchestration. + * + * The in-memory backend does not model orchestration versions or tags, so passing + * either option throws instead of silently diverging from TaskHubGrpcClient. */ async scheduleNewOrchestration( orchestrator: TOrchestrator | string, input?: TInput, instanceId?: string, startAt?: Date, + ): Promise; + async scheduleNewOrchestration( + orchestrator: TOrchestrator | string, + input?: TInput, + options?: StartOrchestrationOptions, + ): Promise; + async scheduleNewOrchestration( + orchestrator: TOrchestrator | string, + input?: TInput, + instanceIdOrOptions?: string | StartOrchestrationOptions, + startAt?: Date, ): Promise { const name = typeof orchestrator === "string" ? orchestrator : getName(orchestrator); + if (typeof instanceIdOrOptions === "object") { + if (instanceIdOrOptions.tags !== undefined) { + throw new Error("TestOrchestrationClient does not support the 'tags' option"); + } + if (instanceIdOrOptions.version !== undefined) { + throw new Error("TestOrchestrationClient does not support the 'version' option"); + } + } + const instanceId = + typeof instanceIdOrOptions === "string" || instanceIdOrOptions === undefined + ? instanceIdOrOptions + : instanceIdOrOptions.instanceId; + const scheduledStartAt = + typeof instanceIdOrOptions === "string" || instanceIdOrOptions === undefined + ? startAt + : instanceIdOrOptions.startAt; + const dedupeStatuses = + typeof instanceIdOrOptions === "string" || instanceIdOrOptions === undefined + ? undefined + : instanceIdOrOptions.dedupeStatuses; const id = instanceId ?? randomUUID(); const encodedInput = input !== undefined ? JSON.stringify(input) : undefined; - this.backend.createInstance(id, name, encodedInput, startAt); + await this.backend.createOrchestrationInstance(id, name, encodedInput, scheduledStartAt, dedupeStatuses); return id; } @@ -150,11 +185,7 @@ export class TestOrchestrationClient { * @param input Optional operation input. Serialized as JSON. */ async signalEntity(id: EntityInstanceId, operationName: string, input?: unknown): Promise { - this.backend.signalEntity( - id.toString(), - operationName, - input === undefined ? undefined : JSON.stringify(input), - ); + this.backend.signalEntity(id.toString(), operationName, input === undefined ? undefined : JSON.stringify(input)); } /** @@ -179,9 +210,7 @@ export class TestOrchestrationClient { lockedBy: entity.lockedBy, includesState: includeState, state: - includeState && entity.serializedState !== undefined - ? (JSON.parse(entity.serializedState) as T) - : undefined, + includeState && entity.serializedState !== undefined ? (JSON.parse(entity.serializedState) as T) : undefined, }; } @@ -196,7 +225,8 @@ export class TestOrchestrationClient { return ( status === pb.OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED || status === pb.OrchestrationStatus.ORCHESTRATION_STATUS_FAILED || - status === pb.OrchestrationStatus.ORCHESTRATION_STATUS_TERMINATED + status === pb.OrchestrationStatus.ORCHESTRATION_STATUS_TERMINATED || + status === pb.OrchestrationStatus.ORCHESTRATION_STATUS_CANCELED ); } diff --git a/packages/durabletask-js/src/testing/test-worker.ts b/packages/durabletask-js/src/testing/test-worker.ts index 857c3283..dbd3b19b 100644 --- a/packages/durabletask-js/src/testing/test-worker.ts +++ b/packages/durabletask-js/src/testing/test-worker.ts @@ -200,15 +200,15 @@ export class TestOrchestrationWorker { * Processes a single activity work item. */ private async processActivity(workItem: ActivityWorkItem): Promise { - const { instanceId, name, taskId, input } = workItem; + const { instanceId, executionId, name, taskId, input } = workItem; try { const executor = new ActivityExecutor(this.registry); const result = await executor.execute(instanceId, name, taskId, input); - this.backend.completeActivity(instanceId, taskId, result); + this.backend.completeActivity(instanceId, executionId, taskId, result); } catch (error: unknown) { const err = error instanceof Error ? error : new Error(String(error)); - this.backend.completeActivity(instanceId, taskId, undefined, err); + this.backend.completeActivity(instanceId, executionId, taskId, undefined, err); } } diff --git a/packages/durabletask-js/test/orchestration-id-reuse-policy.spec.ts b/packages/durabletask-js/test/orchestration-id-reuse-policy.spec.ts new file mode 100644 index 00000000..0da19524 --- /dev/null +++ b/packages/durabletask-js/test/orchestration-id-reuse-policy.spec.ts @@ -0,0 +1,476 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import * as grpc from "@grpc/grpc-js"; +import { + ActivityContext, + InMemoryOrchestrationBackend, + OrchestrationContext, + OrchestrationAlreadyExistsError, + OrchestrationStatus, + TaskHubGrpcClient, + TestOrchestrationClient, + TestOrchestrationWorker, + TOrchestrator, + ValidDedupeStatuses, +} from "../src"; +import * as pb from "../src/proto/orchestrator_service_pb"; +import * as pbh from "../src/utils/pb-helper.util"; + +function deferred(): { + promise: Promise; + resolve: (value: T) => void; +} { + let resolve!: (value: T) => void; + const promise = new Promise((resolvePromise) => { + resolve = resolvePromise; + }); + return { promise, resolve }; +} + +function mockStartInstance( + client: TaskHubGrpcClient, + captureRequest: (request: pb.CreateInstanceRequest) => void, + error?: grpc.ServiceError, +): void { + const stub = (client as unknown as { _stub: Record })._stub; + stub.startInstance = ( + request: pb.CreateInstanceRequest, + _metadata: grpc.Metadata, + callback: (error: grpc.ServiceError | null, response: pb.CreateInstanceResponse) => void, + ) => { + captureRequest(request); + const response = new pb.CreateInstanceResponse(); + response.setInstanceid(request.getInstanceid()); + callback(error ?? null, response); + return {} as grpc.ClientUnaryCall; + }; +} + +describe("TaskHubGrpcClient orchestration ID reuse policy", () => { + let client: TaskHubGrpcClient; + + beforeEach(() => { + client = new TaskHubGrpcClient({ hostAddress: "localhost:4001" }); + }); + + afterEach(async () => { + await client.stop(); + }); + + it("omits the policy by default to preserve backend duplicate-ID behavior", async () => { + let request: pb.CreateInstanceRequest | undefined; + mockStartInstance(client, (value) => { + request = value; + }); + + await client.scheduleNewOrchestration("workflow", undefined, { instanceId: "instance-1" }); + + expect(request?.hasOrchestrationidreusepolicy()).toBe(false); + }); + + it("serializes dedupe statuses as the complement of replaceable statuses", async () => { + let request: pb.CreateInstanceRequest | undefined; + mockStartInstance(client, (value) => { + request = value; + }); + await client.scheduleNewOrchestration("workflow", undefined, { + instanceId: "instance-1", + dedupeStatuses: [OrchestrationStatus.COMPLETED, OrchestrationStatus.FAILED], + }); + + expect(request?.getOrchestrationidreusepolicy()?.getReplaceablestatusList()).toEqual([ + pb.OrchestrationStatus.ORCHESTRATION_STATUS_TERMINATED, + pb.OrchestrationStatus.ORCHESTRATION_STATUS_CANCELED, + pb.OrchestrationStatus.ORCHESTRATION_STATUS_PENDING, + pb.OrchestrationStatus.ORCHESTRATION_STATUS_RUNNING, + pb.OrchestrationStatus.ORCHESTRATION_STATUS_SUSPENDED, + ]); + }); + + it("serializes an empty dedupe list as all replaceable runtime statuses", async () => { + let request: pb.CreateInstanceRequest | undefined; + mockStartInstance(client, (value) => { + request = value; + }); + + await client.scheduleNewOrchestration("workflow", undefined, { + instanceId: "instance-1", + dedupeStatuses: [], + }); + + expect(request?.getOrchestrationidreusepolicy()?.getReplaceablestatusList()).toEqual([ + pb.OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED, + pb.OrchestrationStatus.ORCHESTRATION_STATUS_FAILED, + pb.OrchestrationStatus.ORCHESTRATION_STATUS_TERMINATED, + pb.OrchestrationStatus.ORCHESTRATION_STATUS_CANCELED, + pb.OrchestrationStatus.ORCHESTRATION_STATUS_PENDING, + pb.OrchestrationStatus.ORCHESTRATION_STATUS_RUNNING, + pb.OrchestrationStatus.ORCHESTRATION_STATUS_SUSPENDED, + ]); + }); + + it("exports the same valid dedupe statuses as the .NET SDK", () => { + expect(ValidDedupeStatuses).toEqual([ + OrchestrationStatus.COMPLETED, + OrchestrationStatus.FAILED, + OrchestrationStatus.TERMINATED, + OrchestrationStatus.CANCELED, + OrchestrationStatus.PENDING, + OrchestrationStatus.RUNNING, + OrchestrationStatus.SUSPENDED, + ]); + }); + + it("rejects an invalid dedupe status before calling the sidecar", async () => { + let called = false; + mockStartInstance(client, () => { + called = true; + }); + + await expect( + client.scheduleNewOrchestration("workflow", undefined, { + dedupeStatuses: [999 as OrchestrationStatus], + }), + ).rejects.toThrow(new TypeError("Invalid orchestration runtime status: '999' for deduplication.")); + expect(called).toBe(false); + }); + + it("forwards terminated dedupe with a reusable running status to the sidecar", async () => { + let request: pb.CreateInstanceRequest | undefined; + mockStartInstance(client, (value) => { + request = value; + }); + + await client.scheduleNewOrchestration("workflow", undefined, { + dedupeStatuses: [OrchestrationStatus.TERMINATED, OrchestrationStatus.RUNNING, OrchestrationStatus.PENDING], + }); + + expect(request?.getOrchestrationidreusepolicy()?.getReplaceablestatusList()).toEqual([ + pb.OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED, + pb.OrchestrationStatus.ORCHESTRATION_STATUS_FAILED, + pb.OrchestrationStatus.ORCHESTRATION_STATUS_CANCELED, + pb.OrchestrationStatus.ORCHESTRATION_STATUS_SUSPENDED, + ]); + }); + + it.each([ + [grpc.status.ALREADY_EXISTS, OrchestrationAlreadyExistsError], + [grpc.status.INVALID_ARGUMENT, TypeError], + ])("maps gRPC status %s to the public error contract", async (code, expectedError) => { + const error = Object.assign(new Error("sidecar rejected start"), { + code, + details: "sidecar rejected start", + }) as grpc.ServiceError; + mockStartInstance(client, () => {}, error); + + await expect(client.scheduleNewOrchestration("workflow")).rejects.toBeInstanceOf(expectedError); + }); + + it("preserves a canceled scheduling ServiceError when no public cancellation error exists", async () => { + const error = Object.assign(new Error("schedule canceled"), { + code: grpc.status.CANCELLED, + details: "schedule canceled", + }) as grpc.ServiceError; + mockStartInstance(client, () => {}, error); + + await expect(client.scheduleNewOrchestration("workflow")).rejects.toBe(error); + }); +}); + +describe("TestOrchestrationClient orchestration ID reuse policy", () => { + let backend: InMemoryOrchestrationBackend; + let client: TestOrchestrationClient; + let worker: TestOrchestrationWorker; + + const waitingOrchestrator: TOrchestrator = async function* (ctx: OrchestrationContext): any { + return yield ctx.waitForExternalEvent("finish"); + }; + + beforeEach(async () => { + backend = new InMemoryOrchestrationBackend(); + client = new TestOrchestrationClient(backend); + worker = new TestOrchestrationWorker(backend); + worker.addOrchestrator(waitingOrchestrator); + await worker.start(); + }); + + afterEach(async () => { + await worker.stop(); + backend.reset(); + }); + + it("treats omitted dedupe statuses as all statuses reusable", async () => { + await client.scheduleNewOrchestration(waitingOrchestrator, "original", { + instanceId: "instance-1", + }); + await client.waitForOrchestrationStart("instance-1", false, 5); + const originalExecutionId = backend.getInstance("instance-1")?.executionId; + + await client.scheduleNewOrchestration(waitingOrchestrator, "replacement", { + instanceId: "instance-1", + }); + + const replacement = backend.getInstance("instance-1"); + expect(replacement?.executionId).not.toBe(originalExecutionId); + expect(replacement?.input).toBe(JSON.stringify("replacement")); + }); + + it("rejects reuse when the existing runtime status is selected for deduplication", async () => { + await client.scheduleNewOrchestration(waitingOrchestrator, "original", { + instanceId: "instance-1", + }); + await client.waitForOrchestrationStart("instance-1", false, 5); + + await expect( + client.scheduleNewOrchestration(waitingOrchestrator, "replacement", { + instanceId: "instance-1", + dedupeStatuses: [OrchestrationStatus.RUNNING], + }), + ).rejects.toBeInstanceOf(OrchestrationAlreadyExistsError); + }); + + it("rejects terminated dedupe when a running status remains reusable", async () => { + await expect( + client.scheduleNewOrchestration(waitingOrchestrator, undefined, { + instanceId: "instance-1", + dedupeStatuses: [OrchestrationStatus.TERMINATED, OrchestrationStatus.RUNNING, OrchestrationStatus.PENDING], + }), + ).rejects.toThrow( + new TypeError( + "Invalid dedupe statuses: cannot include 'Terminated' while also allowing reuse of running instances, " + + "because the running instance would be terminated and then immediately conflict with the dedupe check.", + ), + ); + }); + + it("atomically replaces an existing instance whose runtime status is reusable", async () => { + await client.scheduleNewOrchestration(waitingOrchestrator, "original", { + instanceId: "instance-1", + }); + await client.waitForOrchestrationStart("instance-1", false, 5); + const originalExecutionId = backend.getInstance("instance-1")?.executionId; + + await client.scheduleNewOrchestration(waitingOrchestrator, "replacement", { + instanceId: "instance-1", + dedupeStatuses: [ + OrchestrationStatus.COMPLETED, + OrchestrationStatus.FAILED, + OrchestrationStatus.PENDING, + OrchestrationStatus.SUSPENDED, + ], + }); + + const replacement = backend.getInstance("instance-1"); + expect(replacement?.executionId).not.toBe(originalExecutionId); + expect(replacement?.input).toBe(JSON.stringify("replacement")); + expect(replacement?.status).toBe(pb.OrchestrationStatus.ORCHESTRATION_STATUS_PENDING); + }); + + it("terminates the old execution before creating its replacement", async () => { + await client.scheduleNewOrchestration(waitingOrchestrator, "original", { + instanceId: "instance-1", + }); + await client.waitForOrchestrationStart("instance-1", false, 5); + const originalCompletion = client.waitForOrchestrationCompletion("instance-1", true, 5); + + const replacement = client.scheduleNewOrchestration(waitingOrchestrator, "replacement", { + instanceId: "instance-1", + dedupeStatuses: [], + }); + + const originalState = await originalCompletion; + expect(originalState?.runtimeStatus).toBe(OrchestrationStatus.TERMINATED); + await replacement; + }); + + it("maps the canceled protobuf status to the public canceled status", () => { + expect(backend.toClientStatus(pb.OrchestrationStatus.ORCHESTRATION_STATUS_CANCELED)).toBe( + OrchestrationStatus.CANCELED, + ); + }); + + it("treats canceled instances as terminal when waiting for completion", async () => { + const instanceId = await client.scheduleNewOrchestration(waitingOrchestrator, undefined, { + instanceId: "canceled-instance", + }); + await client.waitForOrchestrationStart(instanceId, false, 5); + const instance = backend.getInstance(instanceId)!; + + backend.completeOrchestration(instanceId, instance.completionToken, [ + pbh.newCompleteOrchestrationAction(-1, pb.OrchestrationStatus.ORCHESTRATION_STATUS_CANCELED), + ]); + + const state = await client.waitForOrchestrationCompletion(instanceId, false, 0.1); + expect(state?.runtimeStatus).toBe(OrchestrationStatus.CANCELED); + }); + + it.each([ + ["tags", { tags: { environment: "test" } }], + ["version", { version: "v2" }], + ])("rejects unsupported %s options instead of silently dropping them", async (_name, options) => { + await expect( + client.scheduleNewOrchestration(waitingOrchestrator, undefined, { + instanceId: `unsupported-${_name}`, + ...options, + }), + ).rejects.toThrow(`TestOrchestrationClient does not support the '${_name}' option`); + }); +}); + +describe("TestOrchestrationClient replacement generation fences", () => { + let backend: InMemoryOrchestrationBackend; + let client: TestOrchestrationClient; + let worker: TestOrchestrationWorker; + + beforeEach(() => { + backend = new InMemoryOrchestrationBackend(); + client = new TestOrchestrationClient(backend); + worker = new TestOrchestrationWorker(backend); + }); + + afterEach(async () => { + await worker.stop(); + backend.reset(); + }); + + it("does not deliver a running activity's stale completion to the replacement execution", async () => { + const originalActivityStarted = deferred(); + const releaseOriginalActivity = deferred(); + const activity = async (_ctx: ActivityContext, input: string): Promise => { + if (input === "original") { + originalActivityStarted.resolve(); + await releaseOriginalActivity.promise; + } + return input; + }; + const orchestrator: TOrchestrator = async function* (ctx: OrchestrationContext, input: string): any { + return yield ctx.callActivity(activity, input); + }; + + worker.addOrchestrator(orchestrator); + worker.addActivity(activity); + await worker.start(); + + await client.scheduleNewOrchestration(orchestrator, "original", { instanceId: "activity-race" }); + await originalActivityStarted.promise; + + const replacement = client.scheduleNewOrchestration(orchestrator, "replacement", { + instanceId: "activity-race", + dedupeStatuses: [], + }); + + const replacementProgress = await Promise.race([ + replacement.then(() => "replaced"), + new Promise((resolve) => setTimeout(() => resolve("blocked"), 100)), + ]); + releaseOriginalActivity.resolve(); + + expect(replacementProgress).toBe("replaced"); + await replacement; + + const state = await client.waitForOrchestrationCompletion("activity-race", true, 5); + expect(state?.serializedOutput).toBe(JSON.stringify("replacement")); + }); + + it("does not deliver a stale sub-orchestration completion to the replacement execution", async () => { + const originalChildStarted = deferred(); + const replacementChildStarted = deferred(); + const child: TOrchestrator = async function* (ctx: OrchestrationContext, input: string): any { + if (!ctx.isReplaying) { + (input === "original" ? originalChildStarted : replacementChildStarted).resolve(ctx.instanceId); + } + return yield ctx.waitForExternalEvent("finish"); + }; + const parent: TOrchestrator = async function* (ctx: OrchestrationContext, input: string): any { + return yield ctx.callSubOrchestrator(child, input); + }; + + worker.addOrchestrator(parent); + worker.addOrchestrator(child); + await worker.start(); + + await client.scheduleNewOrchestration(parent, "original", { instanceId: "parent-race" }); + const originalChildId = await originalChildStarted.promise; + + await client.scheduleNewOrchestration(parent, "replacement", { + instanceId: "parent-race", + dedupeStatuses: [], + }); + const replacementChildId = await replacementChildStarted.promise; + + await client.raiseOrchestrationEvent(originalChildId, "finish", "stale"); + await expect(client.waitForOrchestrationCompletion("parent-race", true, 0.1)).rejects.toThrow( + "Timeout waiting for orchestration 'parent-race'", + ); + + await client.raiseOrchestrationEvent(replacementChildId, "finish", "replacement"); + const state = await client.waitForOrchestrationCompletion("parent-race", true, 5); + expect(state?.serializedOutput).toBe(JSON.stringify("replacement")); + }); + + it("fails a current parent when its running child is replaced", async () => { + const childStarted = deferred(); + const child: TOrchestrator = async function* (ctx: OrchestrationContext): any { + if (!ctx.isReplaying) { + childStarted.resolve(); + } + return yield ctx.waitForExternalEvent("finish"); + }; + const parent: TOrchestrator = async function* (ctx: OrchestrationContext): any { + try { + return yield ctx.callSubOrchestrator(child, undefined, { instanceId: "replaced-child" }); + } catch (error: any) { + return `caught: ${error.message}`; + } + }; + + worker.addOrchestrator(parent); + worker.addOrchestrator(child); + await worker.start(); + + await client.scheduleNewOrchestration(parent, undefined, { instanceId: "current-parent" }); + await childStarted.promise; + + await client.scheduleNewOrchestration(child, undefined, { + instanceId: "replaced-child", + dedupeStatuses: [], + }); + + const state = await client.waitForOrchestrationCompletion("current-parent", true, 5); + expect(JSON.parse(state?.serializedOutput ?? "")).toContain("Sub-orchestration failed"); + }); + + it("removes child watchers owned by a replaced parent execution", async () => { + const childStarted = deferred(); + const child: TOrchestrator = async function* (ctx: OrchestrationContext): any { + if (!ctx.isReplaying) { + childStarted.resolve(); + } + return yield ctx.waitForExternalEvent("finish"); + }; + const parent: TOrchestrator = async function* (ctx: OrchestrationContext, input: string): any { + if (input === "replacement") { + return yield ctx.waitForExternalEvent("finish"); + } + return yield ctx.callSubOrchestrator(child, undefined, { instanceId: "orphaned-child" }); + }; + + worker.addOrchestrator(parent); + worker.addOrchestrator(child); + await worker.start(); + + await client.scheduleNewOrchestration(parent, "original", { instanceId: "replaced-parent" }); + await childStarted.promise; + + const stateWaiters = (backend as any).stateWaiters as Map; + expect(stateWaiters.get("orphaned-child")).toHaveLength(1); + + await client.scheduleNewOrchestration(parent, "replacement", { + instanceId: "replaced-parent", + dedupeStatuses: [], + }); + + expect(stateWaiters.has("orphaned-child")).toBe(false); + }); +}); diff --git a/test/e2e-azuremanaged/orchestration-id-reuse-policy.spec.ts b/test/e2e-azuremanaged/orchestration-id-reuse-policy.spec.ts new file mode 100644 index 00000000..aa239008 --- /dev/null +++ b/test/e2e-azuremanaged/orchestration-id-reuse-policy.spec.ts @@ -0,0 +1,140 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { + OrchestrationAlreadyExistsError, + OrchestrationContext, + OrchestrationStatus, + TaskHubGrpcClient, + TaskHubGrpcWorker, + TOrchestrator, +} from "@microsoft/durabletask-js"; +import { + DurableTaskAzureManagedClientBuilder, + DurableTaskAzureManagedWorkerBuilder, +} from "@microsoft/durabletask-js-azuremanaged"; + +const connectionString = process.env.DTS_CONNECTION_STRING; +const endpoint = process.env.ENDPOINT || "localhost:8080"; +const taskHub = process.env.TASKHUB || "default"; + +function createClient(): TaskHubGrpcClient { + return connectionString + ? new DurableTaskAzureManagedClientBuilder().connectionString(connectionString).build() + : new DurableTaskAzureManagedClientBuilder().endpoint(endpoint, taskHub, null).build(); +} + +function createWorker(): TaskHubGrpcWorker { + return connectionString + ? new DurableTaskAzureManagedWorkerBuilder().connectionString(connectionString).build() + : new DurableTaskAzureManagedWorkerBuilder().endpoint(endpoint, taskHub, null).build(); +} + +describe("Orchestration ID reuse policy E2E", () => { + jest.setTimeout(120000); + + let client: TaskHubGrpcClient; + let worker: TaskHubGrpcWorker; + + const reusableOrchestrator: TOrchestrator = async function* ( + ctx: OrchestrationContext, + input: { value: string; wait: boolean }, + ): any { + if (input.wait) { + yield ctx.waitForExternalEvent("finish"); + } + return input.value; + }; + + beforeEach(async () => { + client = createClient(); + worker = createWorker(); + worker.addOrchestrator(reusableOrchestrator); + await worker.start(); + }); + + afterEach(async () => { + await worker.stop(); + await client.stop(); + }); + + it("rejects a duplicate whose running status is selected for deduplication", async () => { + const instanceId = `reuse-dedupe-${Date.now()}`; + await client.scheduleNewOrchestration(reusableOrchestrator, { value: "original", wait: true }, { instanceId }); + await client.waitForOrchestrationStart(instanceId, false, 30); + + await expect( + client.scheduleNewOrchestration( + reusableOrchestrator, + { value: "replacement", wait: false }, + { instanceId, dedupeStatuses: [OrchestrationStatus.RUNNING] }, + ), + ).rejects.toBeInstanceOf(OrchestrationAlreadyExistsError); + + const state = await client.getOrchestrationState(instanceId, true); + expect(state?.serializedInput).toBe(JSON.stringify({ value: "original", wait: true })); + expect(state?.runtimeStatus).toBe(OrchestrationStatus.RUNNING); + }, 60000); + + it("replaces a duplicate whose running status is reusable", async () => { + const instanceId = `reuse-replace-${Date.now()}`; + await client.scheduleNewOrchestration(reusableOrchestrator, { value: "original", wait: true }, { instanceId }); + await client.waitForOrchestrationStart(instanceId, false, 30); + + await client.scheduleNewOrchestration( + reusableOrchestrator, + { value: "replacement", wait: false }, + { instanceId, dedupeStatuses: [] }, + ); + + const state = await client.waitForOrchestrationCompletion(instanceId, true, 30); + expect(state?.serializedInput).toBe(JSON.stringify({ value: "replacement", wait: false })); + expect(state?.serializedOutput).toBe(JSON.stringify("replacement")); + expect(state?.runtimeStatus).toBe(OrchestrationStatus.COMPLETED); + }, 60000); + + it("surfaces the backend rejection for terminated dedupe with reusable suspended instances", async () => { + const instanceId = `reuse-suspended-${Date.now()}`; + await client.scheduleNewOrchestration(reusableOrchestrator, { value: "original", wait: true }, { instanceId }); + await client.waitForOrchestrationStart(instanceId, false, 30); + await client.suspendOrchestration(instanceId); + + let suspendedState = await client.getOrchestrationState(instanceId); + for (let attempt = 0; attempt < 30 && suspendedState?.runtimeStatus !== OrchestrationStatus.SUSPENDED; attempt++) { + await new Promise((resolve) => setTimeout(resolve, 1000)); + suspendedState = await client.getOrchestrationState(instanceId); + } + expect(suspendedState?.runtimeStatus).toBe(OrchestrationStatus.SUSPENDED); + + await expect( + client.scheduleNewOrchestration( + reusableOrchestrator, + { value: "replacement", wait: false }, + { + instanceId, + dedupeStatuses: [OrchestrationStatus.TERMINATED, OrchestrationStatus.RUNNING, OrchestrationStatus.PENDING], + }, + ), + ).rejects.toThrow( + "Invalid reusable statuses: cannot exclude 'Terminated' while also allowing reuse of running instances", + ); + + const state = await client.getOrchestrationState(instanceId, true); + expect(state?.serializedInput).toBe(JSON.stringify({ value: "original", wait: true })); + expect(state?.runtimeStatus).toBe(OrchestrationStatus.SUSPENDED); + }, 120000); + + it("preserves backend default duplicate behavior when dedupe statuses are undefined", async () => { + const instanceId = `reuse-default-${Date.now()}`; + await client.scheduleNewOrchestration(reusableOrchestrator, { value: "original", wait: true }, { instanceId }); + await client.waitForOrchestrationStart(instanceId, false, 30); + + await expect( + client.scheduleNewOrchestration( + reusableOrchestrator, + { value: "replacement", wait: false }, + { instanceId, dedupeStatuses: undefined }, + ), + ).rejects.toBeInstanceOf(OrchestrationAlreadyExistsError); + }, 60000); +});