From bcadb1136feac867867411e7ef672266a7694a85 Mon Sep 17 00:00:00 2001 From: wangbill Date: Wed, 5 Aug 2026 10:14:49 -0700 Subject: [PATCH 1/9] feat: support orchestration ID reuse policy Expose status-based duplicate handling across the core and Azure Functions clients, including faithful in-memory test semantics. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 3 +- README.md | 22 ++ packages/azure-functions-durable/CHANGELOG.md | 1 + packages/azure-functions-durable/README.md | 2 +- .../azure-functions-durable/src/client.ts | 13 +- packages/azure-functions-durable/src/index.ts | 2 + .../src/orchestration-status.ts | 5 +- .../test/unit/client.spec.ts | 24 +++ .../test/unit/query-types.spec.ts | 15 ++ packages/durabletask-js/src/client/client.ts | 13 +- packages/durabletask-js/src/index.ts | 1 + .../enum/orchestration-status.enum.ts | 4 +- .../orchestration-id-reuse-policy.ts | 42 ++++ .../src/task/options/task-options.ts | 6 + .../src/testing/in-memory-backend.ts | 53 ++++- .../durabletask-js/src/testing/test-client.ts | 35 ++- .../orchestration-id-reuse-policy.spec.ts | 200 ++++++++++++++++++ 17 files changed, 425 insertions(+), 16 deletions(-) create mode 100644 packages/durabletask-js/src/orchestration/orchestration-id-reuse-policy.ts create mode 100644 packages/durabletask-js/test/orchestration-id-reuse-policy.spec.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 3eca2ca9..799886c9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,10 +3,11 @@ ### New - Implement entity support in the in-memory testing backend ([#341](https://github.com/microsoft/durabletask-js/pull/341)) +- Add orchestration instance ID reuse policies with status-based deduplication and atomic replacement. +- Add the `CANCELED` member to the public `OrchestrationStatus` enum. ### Fixes - ## v0.4.0 (2026-07-31) ### Changes diff --git a/README.md b/README.md index 683638b2..5cebdedf 100644 --- a/README.md +++ b/README.md @@ -90,6 +90,28 @@ console.log(`Result: ${state?.serializedOutput}`); You can find more samples in the [examples/azure-managed](./examples/azure-managed) directory. +### Reusing orchestration instance IDs + +Pass an `orchestrationIdReusePolicy` when an instance ID may be reused. `dedupeStatuses` +lists the existing runtime statuses that must continue to produce a duplicate-ID error; +instances in every other runtime status are atomically replaced: + +```typescript +import { OrchestrationStatus } from "@microsoft/durabletask-js"; + +await client.scheduleNewOrchestration(helloCities, undefined, { + instanceId: "daily-greeting", + orchestrationIdReusePolicy: { + dedupeStatuses: [OrchestrationStatus.RUNNING, OrchestrationStatus.PENDING], + }, +}); +``` + +Omitting the policy preserves the backend's default duplicate-ID behavior. An empty +`dedupeStatuses` list makes every runtime status replaceable. 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. diff --git a/packages/azure-functions-durable/CHANGELOG.md b/packages/azure-functions-durable/CHANGELOG.md index 05fc47f7..e9df0ffd 100644 --- a/packages/azure-functions-durable/CHANGELOG.md +++ b/packages/azure-functions-durable/CHANGELOG.md @@ -8,6 +8,7 @@ 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 orchestration instance ID reuse policies through `DurableFunctionsClient.startNew()`. ### Fixes diff --git a/packages/azure-functions-durable/README.md b/packages/azure-functions-durable/README.md index 19694078..b1450482 100644 --- a/packages/azure-functions-durable/README.md +++ b/packages/azure-functions-durable/README.md @@ -37,7 +37,7 @@ 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 `orchestrationIdReusePolicy` options.** - **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..6ec49069 100644 --- a/packages/azure-functions-durable/src/client.ts +++ b/packages/azure-functions-durable/src/client.ts @@ -6,6 +6,7 @@ import { status as grpcStatus } from "@grpc/grpc-js"; import { EntityInstanceId, OrchestrationQuery, + OrchestrationIdReusePolicy, OrchestrationState, OrchestrationStatus, PurgeInstanceCriteria, @@ -73,6 +74,8 @@ export interface StartNewOptions { instanceId?: string; /** Orchestration version to assign (forwarded to the core scheduler). */ version?: string; + /** Controls duplicate instance-ID handling based on the existing orchestration's runtime status. */ + orchestrationIdReusePolicy?: OrchestrationIdReusePolicy; } /** @@ -213,8 +216,14 @@ 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?.orchestrationIdReusePolicy !== undefined + ? { + instanceId: options?.instanceId, + version: options?.version, + orchestrationIdReusePolicy: options?.orchestrationIdReusePolicy, + } : 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..236edac7 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,7 @@ export { createAzureFunctionsMetadataGenerator } from "./metadata"; export { DurableFunctionsWorker } from "./worker"; export { DurableBindingMetadata, addDurableGrpcMetadata } from "./durable-grpc"; export { RetryOptions } from "./retry-options"; +export { OrchestrationIdReusePolicy, StartOrchestrationOptions } 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..c12369ae 100644 --- a/packages/azure-functions-durable/test/unit/client.spec.ts +++ b/packages/azure-functions-durable/test/unit/client.spec.ts @@ -4,6 +4,7 @@ import { HttpRequest } from "@azure/functions"; import { status as grpcStatus } from "@grpc/grpc-js"; import { + OrchestrationIdReusePolicy, OrchestrationState, OrchestrationStatus, PurgeInstanceCriteria, @@ -384,6 +385,29 @@ describe("DurableFunctionsClient", () => { await client.stop(); } }); + + it("startNew forwards the orchestration ID reuse policy to the core scheduler", async () => { + const client = new DurableFunctionsClient(CLIENT_CONFIG); + try { + const schedule = jest.spyOn(client, "scheduleNewOrchestration").mockResolvedValue("inst-9"); + const orchestrationIdReusePolicy: OrchestrationIdReusePolicy = { + dedupeStatuses: [OrchestrationStatus.RUNNING], + }; + + await client.startNew("MyOrch", { + instanceId: "inst-9", + orchestrationIdReusePolicy, + }); + + expect(schedule).toHaveBeenCalledWith("MyOrch", undefined, { + instanceId: "inst-9", + version: undefined, + orchestrationIdReusePolicy, + }); + } finally { + await client.stop(); + } + }); }); describe("control-plane error mapping (v3 parity)", () => { 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/src/client/client.ts b/packages/durabletask-js/src/client/client.ts index af90fc7c..7853e039 100644 --- a/packages/durabletask-js/src/client/client.ts +++ b/packages/durabletask-js/src/client/client.ts @@ -27,6 +27,7 @@ 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 { mapToRecord } from "../utils/tags.util"; import { populateTagsMap } from "../utils/pb-helper.util"; import { EntityInstanceId } from "../entities/entity-instance-id"; @@ -174,7 +175,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 +212,10 @@ export class TaskHubGrpcClient { typeof instanceIdOrOptions === "string" || instanceIdOrOptions === undefined ? undefined : instanceIdOrOptions.version; + const orchestrationIdReusePolicy = + typeof instanceIdOrOptions === "string" || instanceIdOrOptions === undefined + ? undefined + : instanceIdOrOptions.orchestrationIdReusePolicy; // Use provided version, or fall back to client's default version const effectiveVersion = version ?? this._defaultVersion; @@ -234,6 +239,10 @@ export class TaskHubGrpcClient { req.setVersion(v); } + if (orchestrationIdReusePolicy) { + req.setOrchestrationidreusepolicy(toProtobufOrchestrationIdReusePolicy(orchestrationIdReusePolicy)); + } + populateTagsMap(req.getTagsMap(), tags); // Create a tracing span for the new orchestration (if OTEL is available) @@ -1299,4 +1308,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..06b83e80 100644 --- a/packages/durabletask-js/src/index.ts +++ b/packages/durabletask-js/src/index.ts @@ -28,6 +28,7 @@ export { TERMINATE_OPTIONS_SYMBOL, } from "./orchestration/orchestration-terminate-options"; export { OrchestrationStatus } from "./orchestration/enum/orchestration-status.enum"; +export { OrchestrationIdReusePolicy } from "./orchestration/orchestration-id-reuse-policy"; 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/orchestration-id-reuse-policy.ts b/packages/durabletask-js/src/orchestration/orchestration-id-reuse-policy.ts new file mode 100644 index 00000000..16f89166 --- /dev/null +++ b/packages/durabletask-js/src/orchestration/orchestration-id-reuse-policy.ts @@ -0,0 +1,42 @@ +// 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"; + +const REPLACEABLE_RUNTIME_STATUSES: readonly OrchestrationStatus[] = [ + OrchestrationStatus.RUNNING, + OrchestrationStatus.COMPLETED, + OrchestrationStatus.FAILED, + OrchestrationStatus.CANCELED, + OrchestrationStatus.TERMINATED, + OrchestrationStatus.PENDING, + OrchestrationStatus.SUSPENDED, +]; + +/** + * Controls what happens when a new orchestration uses an existing instance ID. + */ +export interface OrchestrationIdReusePolicy { + /** + * Existing instances in one of these runtime statuses must not be replaced. + * + * An empty list makes every runtime status replaceable. Omitting the policy + * preserves the backend's default duplicate-ID behavior. + */ + readonly dedupeStatuses: readonly OrchestrationStatus[]; +} + +/** @hidden Converts the public deduplication policy to the wire-level replacement policy. */ +export function toProtobufOrchestrationIdReusePolicy( + policy: OrchestrationIdReusePolicy, +): pb.OrchestrationIdReusePolicy { + const dedupeStatuses = new Set(policy.dedupeStatuses.map(toProtobuf)); + const replaceableStatuses = REPLACEABLE_RUNTIME_STATUSES.map(toProtobuf).filter( + (status) => !dedupeStatuses.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..3838e1ea 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 { OrchestrationIdReusePolicy } from "../../orchestration/orchestration-id-reuse-policy"; /** * Union type representing the available retry strategies for a task. @@ -74,6 +75,11 @@ export interface StartOrchestrationOptions { * via the OrchestrationContext.version property. */ version?: string; + /** + * Controls whether an existing orchestration with the same instance ID is + * deduplicated or atomically replaced based on its runtime status. + */ + orchestrationIdReusePolicy?: OrchestrationIdReusePolicy; } /** diff --git a/packages/durabletask-js/src/testing/in-memory-backend.ts b/packages/durabletask-js/src/testing/in-memory-backend.ts index a05cde63..59be8862 100644 --- a/packages/durabletask-js/src/testing/in-memory-backend.ts +++ b/packages/durabletask-js/src/testing/in-memory-backend.ts @@ -7,6 +7,7 @@ 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 { OrchestrationIdReusePolicy } from "../orchestration/orchestration-id-reuse-policy"; /** Mints a fresh per-execution ID (DTFx `Guid.ToString("N")` idiom: 32 hex chars, no dashes). */ function newExecutionId(): string { @@ -102,12 +103,12 @@ interface StateWaiter { /** * 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. @@ -148,9 +149,16 @@ export class InMemoryOrchestrationBackend { input?: string, scheduledStartTime?: Date, parentInstance?: ParentOrchestrationInstance, + orchestrationIdReusePolicy?: OrchestrationIdReusePolicy, ): string { - if (this.instances.has(instanceId)) { - throw new Error(`Orchestration instance '${instanceId}' already exists`); + const existingInstance = this.instances.get(instanceId); + if (existingInstance) { + const existingStatus = this.toClientStatus(existingInstance.status); + if (!orchestrationIdReusePolicy || orchestrationIdReusePolicy.dedupeStatuses.includes(existingStatus)) { + throw new Error(`Orchestration instance '${instanceId}' already exists`); + } + + this.removeInstanceForReplacement(instanceId); } const now = new Date(); @@ -362,7 +370,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; } @@ -777,6 +785,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,6 +806,39 @@ export class InMemoryOrchestrationBackend { } } + private removeInstanceForReplacement(instanceId: string): void { + this.cancelInstanceTimers(instanceId); + this.rejectStateWaiters( + 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 rejectStateWaiters(instanceId: string, error: Error): void { + const waiters = this.stateWaiters.get(instanceId); + if (!waiters) { + return; + } + + this.stateWaiters.delete(instanceId); + for (const waiter of waiters) { + waiter.reject(error); + } + } + private isTerminalStatus(status: pb.OrchestrationStatus): boolean { return ( status === pb.OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED || diff --git a/packages/durabletask-js/src/testing/test-client.ts b/packages/durabletask-js/src/testing/test-client.ts index 517fbca2..a567ce0a 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. */ @@ -29,12 +30,42 @@ export class TestOrchestrationClient { 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); + const instanceId = + typeof instanceIdOrOptions === "string" || instanceIdOrOptions === undefined + ? instanceIdOrOptions + : instanceIdOrOptions.instanceId; + const scheduledStartAt = + typeof instanceIdOrOptions === "string" || instanceIdOrOptions === undefined + ? startAt + : instanceIdOrOptions.startAt; + const orchestrationIdReusePolicy = + typeof instanceIdOrOptions === "string" || instanceIdOrOptions === undefined + ? undefined + : instanceIdOrOptions.orchestrationIdReusePolicy; const id = instanceId ?? randomUUID(); const encodedInput = input !== undefined ? JSON.stringify(input) : undefined; - this.backend.createInstance(id, name, encodedInput, startAt); + this.backend.createInstance( + id, + name, + encodedInput, + scheduledStartAt, + undefined, + orchestrationIdReusePolicy, + ); return id; } 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..b1d03adf --- /dev/null +++ b/packages/durabletask-js/test/orchestration-id-reuse-policy.spec.ts @@ -0,0 +1,200 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import * as grpc from "@grpc/grpc-js"; +import { + InMemoryOrchestrationBackend, + OrchestrationContext, + OrchestrationIdReusePolicy, + OrchestrationStatus, + TaskHubGrpcClient, + TestOrchestrationClient, + TestOrchestrationWorker, + TOrchestrator, +} from "../src"; +import * as pb from "../src/proto/orchestrator_service_pb"; + +function mockStartInstance( + client: TaskHubGrpcClient, + captureRequest: (request: pb.CreateInstanceRequest) => void, +): 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(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; + }); + const reusePolicy: OrchestrationIdReusePolicy = { + dedupeStatuses: [OrchestrationStatus.COMPLETED, OrchestrationStatus.FAILED], + }; + + await client.scheduleNewOrchestration("workflow", undefined, { + instanceId: "instance-1", + orchestrationIdReusePolicy: reusePolicy, + }); + + expect(request?.getOrchestrationidreusepolicy()?.getReplaceablestatusList()).toEqual([ + pb.OrchestrationStatus.ORCHESTRATION_STATUS_RUNNING, + pb.OrchestrationStatus.ORCHESTRATION_STATUS_CANCELED, + pb.OrchestrationStatus.ORCHESTRATION_STATUS_TERMINATED, + pb.OrchestrationStatus.ORCHESTRATION_STATUS_PENDING, + 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", + orchestrationIdReusePolicy: { dedupeStatuses: [] }, + }); + + expect(request?.getOrchestrationidreusepolicy()?.getReplaceablestatusList()).toEqual([ + pb.OrchestrationStatus.ORCHESTRATION_STATUS_RUNNING, + pb.OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED, + pb.OrchestrationStatus.ORCHESTRATION_STATUS_FAILED, + pb.OrchestrationStatus.ORCHESTRATION_STATUS_CANCELED, + pb.OrchestrationStatus.ORCHESTRATION_STATUS_TERMINATED, + pb.OrchestrationStatus.ORCHESTRATION_STATUS_PENDING, + pb.OrchestrationStatus.ORCHESTRATION_STATUS_SUSPENDED, + ]); + }); +}); + +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("preserves the default duplicate-ID error behavior", async () => { + await client.scheduleNewOrchestration(waitingOrchestrator, "original", { + instanceId: "instance-1", + }); + + await expect( + client.scheduleNewOrchestration(waitingOrchestrator, "replacement", { + instanceId: "instance-1", + }), + ).rejects.toThrow("already exists"); + }); + + 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", + orchestrationIdReusePolicy: { + dedupeStatuses: [OrchestrationStatus.RUNNING], + }, + }), + ).rejects.toThrow("already exists"); + }); + + 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", + orchestrationIdReusePolicy: { + dedupeStatuses: [ + OrchestrationStatus.COMPLETED, + OrchestrationStatus.FAILED, + OrchestrationStatus.TERMINATED, + 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("rejects waiters that were registered for the replaced execution", 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 originalCompletionAssertion = expect(originalCompletion).rejects.toThrow("was replaced by a new execution"); + + await client.scheduleNewOrchestration(waitingOrchestrator, "replacement", { + instanceId: "instance-1", + orchestrationIdReusePolicy: { dedupeStatuses: [] }, + }); + + await originalCompletionAssertion; + }); + + it("maps the canceled protobuf status to the public canceled status", () => { + expect(backend.toClientStatus(pb.OrchestrationStatus.ORCHESTRATION_STATUS_CANCELED)).toBe( + OrchestrationStatus.CANCELED, + ); + }); +}); From 25f00a5635f62f83eb91b65b8d2ca6cbab08249e Mon Sep 17 00:00:00 2001 From: wangbill Date: Wed, 5 Aug 2026 11:06:24 -0700 Subject: [PATCH 2/9] fix: fence in-memory orchestration replacements Reject unsupported test-client scheduling metadata, document the protocol's missing atomic IGNORE action, and keep canceled status terminal across testing and export-history surfaces. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 3 +- README.md | 9 +- packages/azure-functions-durable/CHANGELOG.md | 3 +- packages/azure-functions-durable/README.md | 4 +- .../azure-functions-durable/src/client.ts | 5 +- .../export-instance-history-activity.ts | 1 + .../src/models/export-job-creation-options.ts | 17 +-- .../test/models.spec.ts | 1 + .../orchestration-id-reuse-policy.ts | 11 +- .../src/testing/in-memory-backend.ts | 29 +++- .../durabletask-js/src/testing/test-client.ts | 14 +- .../durabletask-js/src/testing/test-worker.ts | 6 +- .../orchestration-id-reuse-policy.spec.ts | 124 ++++++++++++++++++ 13 files changed, 193 insertions(+), 34 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 799886c9..9fe62f21 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,7 +3,8 @@ ### New - Implement entity support in the in-memory testing backend ([#341](https://github.com/microsoft/durabletask-js/pull/341)) -- Add orchestration instance ID reuse policies with status-based deduplication and atomic replacement. +- Add orchestration instance ID reuse policies with status-based duplicate rejection and atomic + replacement. The current shared protocol does not support an atomic no-op/`IGNORE` action. - Add the `CANCELED` member to the public `OrchestrationStatus` enum. ### Fixes diff --git a/README.md b/README.md index 5cebdedf..96f22007 100644 --- a/README.md +++ b/README.md @@ -94,7 +94,7 @@ You can find more samples in the [examples/azure-managed](./examples/azure-manag Pass an `orchestrationIdReusePolicy` when an instance ID may be reused. `dedupeStatuses` lists the existing runtime statuses that must continue to produce a duplicate-ID error; -instances in every other runtime status are atomically replaced: +instances in every other supported runtime status are atomically replaced: ```typescript import { OrchestrationStatus } from "@microsoft/durabletask-js"; @@ -108,9 +108,10 @@ await client.scheduleNewOrchestration(helloCities, undefined, { ``` Omitting the policy preserves the backend's default duplicate-ID behavior. An empty -`dedupeStatuses` list makes every runtime status replaceable. 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. +`dedupeStatuses` list makes every supported runtime status replaceable. The transient +`CONTINUED_AS_NEW` status is not replaceable. 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 diff --git a/packages/azure-functions-durable/CHANGELOG.md b/packages/azure-functions-durable/CHANGELOG.md index e9df0ffd..9f2c5729 100644 --- a/packages/azure-functions-durable/CHANGELOG.md +++ b/packages/azure-functions-durable/CHANGELOG.md @@ -8,7 +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 orchestration instance ID reuse policies through `DurableFunctionsClient.startNew()`. +- Forward status-based duplicate rejection and atomic replacement policies 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 b1450482..1517f356 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` and `orchestrationIdReusePolicy` options.** + **`client.startNew()` supports the `version` and `orchestrationIdReusePolicy` options.** The reuse + policy supports status-based duplicate errors and atomic replacement; 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 6ec49069..087e8867 100644 --- a/packages/azure-functions-durable/src/client.ts +++ b/packages/azure-functions-durable/src/client.ts @@ -74,7 +74,10 @@ export interface StartNewOptions { instanceId?: string; /** Orchestration version to assign (forwarded to the core scheduler). */ version?: string; - /** Controls duplicate instance-ID handling based on the existing orchestration's runtime status. */ + /** + * Controls duplicate-ID errors and atomic replacement by existing runtime status. + * The current shared protocol does not support an atomic no-op/IGNORE action. + */ orchestrationIdReusePolicy?: OrchestrationIdReusePolicy; } 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/orchestration/orchestration-id-reuse-policy.ts b/packages/durabletask-js/src/orchestration/orchestration-id-reuse-policy.ts index 16f89166..ec95d78e 100644 --- a/packages/durabletask-js/src/orchestration/orchestration-id-reuse-policy.ts +++ b/packages/durabletask-js/src/orchestration/orchestration-id-reuse-policy.ts @@ -15,14 +15,17 @@ const REPLACEABLE_RUNTIME_STATUSES: readonly OrchestrationStatus[] = [ ]; /** - * Controls what happens when a new orchestration uses an existing instance ID. + * Controls whether a new orchestration rejects or atomically replaces an existing instance. + * + * The current shared protocol does not support an atomic no-op/IGNORE action. */ export interface OrchestrationIdReusePolicy { /** - * Existing instances in one of these runtime statuses must not be replaced. + * Existing instances in one of these runtime statuses produce a duplicate-ID error. * - * An empty list makes every runtime status replaceable. Omitting the policy - * preserves the backend's default duplicate-ID behavior. + * Instances in every other supported runtime status are atomically replaced. An empty + * list makes every supported runtime status replaceable. Omitting the policy preserves the + * backend's default duplicate-ID behavior. */ readonly 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 59be8862..a2d34358 100644 --- a/packages/durabletask-js/src/testing/in-memory-backend.ts +++ b/packages/durabletask-js/src/testing/in-memory-backend.ts @@ -38,6 +38,7 @@ export interface OrchestrationInstance { */ export interface ActivityWorkItem { instanceId: string; + executionId: string; name: string; taskId: number; input?: string; @@ -648,6 +649,7 @@ export class InMemoryOrchestrationBackend { */ completeActivity( instanceId: string, + executionId: string, taskId: number, result?: string, error?: Error, @@ -656,6 +658,9 @@ export class InMemoryOrchestrationBackend { 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) { @@ -843,7 +848,8 @@ export class InMemoryOrchestrationBackend { 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 ); } @@ -954,6 +960,7 @@ export class InMemoryOrchestrationBackend { // Queue activity for execution this.activityQueue.push({ instanceId: instance.instanceId, + executionId: instance.executionId, name: taskName, taskId, input, @@ -978,12 +985,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(); @@ -1019,7 +1031,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)); @@ -1029,7 +1041,12 @@ 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( @@ -1041,7 +1058,7 @@ export class InMemoryOrchestrationBackend { 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; } @@ -1186,7 +1203,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 diff --git a/packages/durabletask-js/src/testing/test-client.ts b/packages/durabletask-js/src/testing/test-client.ts index a567ce0a..18aea395 100644 --- a/packages/durabletask-js/src/testing/test-client.ts +++ b/packages/durabletask-js/src/testing/test-client.ts @@ -24,6 +24,9 @@ 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, @@ -43,6 +46,14 @@ export class TestOrchestrationClient { 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 @@ -227,7 +238,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 index b1d03adf..7d3d7fc9 100644 --- a/packages/durabletask-js/test/orchestration-id-reuse-policy.spec.ts +++ b/packages/durabletask-js/test/orchestration-id-reuse-policy.spec.ts @@ -3,6 +3,7 @@ import * as grpc from "@grpc/grpc-js"; import { + ActivityContext, InMemoryOrchestrationBackend, OrchestrationContext, OrchestrationIdReusePolicy, @@ -13,6 +14,18 @@ import { TOrchestrator, } 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, @@ -197,4 +210,115 @@ describe("TestOrchestrationClient orchestration ID reuse policy", () => { 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; + + await client.scheduleNewOrchestration(orchestrator, "replacement", { + instanceId: "activity-race", + orchestrationIdReusePolicy: { dedupeStatuses: [] }, + }); + releaseOriginalActivity.resolve(); + + 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", + orchestrationIdReusePolicy: { 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")); + }); }); From 13f420863301b00f5ed072d6c5f280c028e44630 Mon Sep 17 00:00:00 2001 From: wangbill Date: Wed, 5 Aug 2026 11:45:55 -0700 Subject: [PATCH 3/9] fix: track sub-orchestration watcher ownership Notify a live parent when its child is replaced and cancel child-keyed watchers when their owning parent execution is replaced. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 6157e6cd-9ce9-4636-a8d5-4f49f4f32de8 --- .../src/testing/in-memory-backend.ts | 86 +++++++++++++++++-- .../orchestration-id-reuse-policy.spec.ts | 67 +++++++++++++++ 2 files changed, 148 insertions(+), 5 deletions(-) diff --git a/packages/durabletask-js/src/testing/in-memory-backend.ts b/packages/durabletask-js/src/testing/in-memory-backend.ts index a2d34358..e52132bb 100644 --- a/packages/durabletask-js/src/testing/in-memory-backend.ts +++ b/packages/durabletask-js/src/testing/in-memory-backend.ts @@ -100,6 +100,13 @@ 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; } /** @@ -681,6 +688,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)) { @@ -689,7 +705,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. @@ -711,6 +727,7 @@ export class InMemoryOrchestrationBackend { reject(error); }, predicate, + subOrchestrationWatcher, }; if (timeoutMs > 0) { @@ -812,8 +829,18 @@ export class InMemoryOrchestrationBackend { } private removeInstanceForReplacement(instanceId: string): void { + const instance = this.instances.get(instanceId); + if (!instance) { + return; + } + this.cancelInstanceTimers(instanceId); - this.rejectStateWaiters( + 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`), ); @@ -832,7 +859,7 @@ export class InMemoryOrchestrationBackend { } } - private rejectStateWaiters(instanceId: string, error: Error): void { + private rejectStateWaitersForReplacement(instanceId: string, error: Error): void { const waiters = this.stateWaiters.get(instanceId); if (!waiters) { return; @@ -840,10 +867,58 @@ export class InMemoryOrchestrationBackend { 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 isTerminalStatus(status: pb.OrchestrationStatus): boolean { return ( status === pb.OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED || @@ -1049,10 +1124,11 @@ export class InMemoryOrchestrationBackend { ): 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); @@ -1080,7 +1156,7 @@ 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. }); } diff --git a/packages/durabletask-js/test/orchestration-id-reuse-policy.spec.ts b/packages/durabletask-js/test/orchestration-id-reuse-policy.spec.ts index 7d3d7fc9..2dd16db8 100644 --- a/packages/durabletask-js/test/orchestration-id-reuse-policy.spec.ts +++ b/packages/durabletask-js/test/orchestration-id-reuse-policy.spec.ts @@ -321,4 +321,71 @@ describe("TestOrchestrationClient replacement generation fences", () => { 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", + orchestrationIdReusePolicy: { dedupeStatuses: [] }, + }); + + const state = await client.waitForOrchestrationCompletion("current-parent", true, 5); + expect(JSON.parse(state?.serializedOutput ?? "")).toContain( + "Sub-orchestration instance 'replaced-child' was replaced by a new execution", + ); + }); + + 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", + orchestrationIdReusePolicy: { dedupeStatuses: [] }, + }); + + expect(stateWaiters.has("orphaned-child")).toBe(false); + }); }); From c751fc01c4a83973cbc7ddd21ab46f95f9231328 Mon Sep 17 00:00:00 2001 From: wangbill Date: Thu, 6 Aug 2026 13:09:40 -0700 Subject: [PATCH 4/9] feat: align orchestration ID reuse with .NET Expose flat dedupe statuses, validate and serialize the status complement, map duplicate errors, and align in-memory replacement semantics. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: d491ec0d-9ce9-421b-9953-7d179d53625b --- CHANGELOG.md | 7 +- README.md | 42 ++---- packages/azure-functions-durable/CHANGELOG.md | 2 +- packages/azure-functions-durable/README.md | 6 +- .../azure-functions-durable/src/client.ts | 11 +- packages/azure-functions-durable/src/index.ts | 6 +- .../test/unit/client.spec.ts | 20 +-- packages/durabletask-js/src/client/client.ts | 57 +++++-- packages/durabletask-js/src/index.ts | 3 +- .../orchestration-already-exists-error.ts | 10 ++ .../orchestration-id-reuse-policy.ts | 54 ++++--- .../src/task/options/task-options.ts | 10 +- .../src/testing/in-memory-backend.ts | 142 ++++++++++++------ .../durabletask-js/src/testing/test-client.ts | 23 +-- .../orchestration-id-reuse-policy.spec.ts | 136 ++++++++++++----- .../orchestration-id-reuse-policy.spec.ts | 94 ++++++++++++ 16 files changed, 437 insertions(+), 186 deletions(-) create mode 100644 packages/durabletask-js/src/orchestration/exception/orchestration-already-exists-error.ts create mode 100644 test/e2e-azuremanaged/orchestration-id-reuse-policy.spec.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 9fe62f21..5f983eb7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,8 +3,9 @@ ### New - Implement entity support in the in-memory testing backend ([#341](https://github.com/microsoft/durabletask-js/pull/341)) -- Add orchestration instance ID reuse policies with status-based duplicate rejection and atomic - replacement. The current shared protocol does not support an atomic no-op/`IGNORE` action. +- Add the top-level `StartOrchestrationOptions.dedupeStatuses` option, `ValidDedupeStatuses`, and + `OrchestrationAlreadyExistsError`, aligned with the .NET status-based duplicate rejection and + atomic replacement contract. The shared protocol does not support atomic no-op/`IGNORE`. - Add the `CANCELED` member to the public `OrchestrationStatus` enum. ### Fixes @@ -74,7 +75,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 96f22007..d9592b30 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 => { @@ -92,8 +85,9 @@ You can find more samples in the [examples/azure-managed](./examples/azure-manag ### Reusing orchestration instance IDs -Pass an `orchestrationIdReusePolicy` when an instance ID may be reused. `dedupeStatuses` -lists the existing runtime statuses that must continue to produce a duplicate-ID error; +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 @@ -101,17 +95,16 @@ import { OrchestrationStatus } from "@microsoft/durabletask-js"; await client.scheduleNewOrchestration(helloCities, undefined, { instanceId: "daily-greeting", - orchestrationIdReusePolicy: { - dedupeStatuses: [OrchestrationStatus.RUNNING, OrchestrationStatus.PENDING], - }, + dedupeStatuses: [OrchestrationStatus.RUNNING, OrchestrationStatus.PENDING], }); ``` -Omitting the policy preserves the backend's default duplicate-ID behavior. An empty -`dedupeStatuses` list makes every supported runtime status replaceable. The transient -`CONTINUED_AS_NEW` status is not replaceable. 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. +Omitting `dedupeStatuses` preserves the backend's default duplicate-ID behavior; passing `[]` +makes every supported runtime status replaceable. `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 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 @@ -148,10 +141,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 9f2c5729..880f1bd6 100644 --- a/packages/azure-functions-durable/CHANGELOG.md +++ b/packages/azure-functions-durable/CHANGELOG.md @@ -8,7 +8,7 @@ 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 status-based duplicate rejection and atomic replacement policies through +- 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 1517f356..d672a302 100644 --- a/packages/azure-functions-durable/README.md +++ b/packages/azure-functions-durable/README.md @@ -37,9 +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` and `orchestrationIdReusePolicy` options.** The reuse - policy supports status-based duplicate errors and atomic replacement; the current shared protocol - does not expose an atomic no-op/`IGNORE` action. + **`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 087e8867..37cdb53c 100644 --- a/packages/azure-functions-durable/src/client.ts +++ b/packages/azure-functions-durable/src/client.ts @@ -6,7 +6,6 @@ import { status as grpcStatus } from "@grpc/grpc-js"; import { EntityInstanceId, OrchestrationQuery, - OrchestrationIdReusePolicy, OrchestrationState, OrchestrationStatus, PurgeInstanceCriteria, @@ -75,10 +74,10 @@ export interface StartNewOptions { /** Orchestration version to assign (forwarded to the core scheduler). */ version?: string; /** - * Controls duplicate-ID errors and atomic replacement by existing runtime status. + * Existing orchestration statuses that must produce a duplicate-ID error. * The current shared protocol does not support an atomic no-op/IGNORE action. */ - orchestrationIdReusePolicy?: OrchestrationIdReusePolicy; + dedupeStatuses?: readonly OrchestrationStatus[]; } /** @@ -219,13 +218,11 @@ export class DurableFunctionsClient extends TaskHubGrpcClient { */ async startNew(orchestratorName: string, options?: StartNewOptions): Promise { const scheduleOptions = - options?.instanceId !== undefined || - options?.version !== undefined || - options?.orchestrationIdReusePolicy !== undefined + options?.instanceId !== undefined || options?.version !== undefined || options?.dedupeStatuses !== undefined ? { instanceId: options?.instanceId, version: options?.version, - orchestrationIdReusePolicy: options?.orchestrationIdReusePolicy, + 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 236edac7..81862725 100644 --- a/packages/azure-functions-durable/src/index.ts +++ b/packages/azure-functions-durable/src/index.ts @@ -23,7 +23,11 @@ export { createAzureFunctionsMetadataGenerator } from "./metadata"; export { DurableFunctionsWorker } from "./worker"; export { DurableBindingMetadata, addDurableGrpcMetadata } from "./durable-grpc"; export { RetryOptions } from "./retry-options"; -export { OrchestrationIdReusePolicy, StartOrchestrationOptions } from "@microsoft/durabletask-js"; +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/test/unit/client.spec.ts b/packages/azure-functions-durable/test/unit/client.spec.ts index c12369ae..7bebfeda 100644 --- a/packages/azure-functions-durable/test/unit/client.spec.ts +++ b/packages/azure-functions-durable/test/unit/client.spec.ts @@ -4,7 +4,6 @@ import { HttpRequest } from "@azure/functions"; import { status as grpcStatus } from "@grpc/grpc-js"; import { - OrchestrationIdReusePolicy, OrchestrationState, OrchestrationStatus, PurgeInstanceCriteria, @@ -386,23 +385,21 @@ describe("DurableFunctionsClient", () => { } }); - it("startNew forwards the orchestration ID reuse policy to the core scheduler", async () => { + 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 orchestrationIdReusePolicy: OrchestrationIdReusePolicy = { - dedupeStatuses: [OrchestrationStatus.RUNNING], - }; + const dedupeStatuses = [OrchestrationStatus.RUNNING]; await client.startNew("MyOrch", { instanceId: "inst-9", - orchestrationIdReusePolicy, + dedupeStatuses, }); expect(schedule).toHaveBeenCalledWith("MyOrch", undefined, { instanceId: "inst-9", version: undefined, - orchestrationIdReusePolicy, + dedupeStatuses, }); } finally { await client.stop(); @@ -427,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. @@ -567,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/durabletask-js/src/client/client.ts b/packages/durabletask-js/src/client/client.ts index 7853e039..0d5f2072 100644 --- a/packages/durabletask-js/src/client/client.ts +++ b/packages/durabletask-js/src/client/client.ts @@ -28,6 +28,7 @@ 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"; @@ -212,10 +213,10 @@ export class TaskHubGrpcClient { typeof instanceIdOrOptions === "string" || instanceIdOrOptions === undefined ? undefined : instanceIdOrOptions.version; - const orchestrationIdReusePolicy = + const dedupeStatuses = typeof instanceIdOrOptions === "string" || instanceIdOrOptions === undefined ? undefined - : instanceIdOrOptions.orchestrationIdReusePolicy; + : instanceIdOrOptions.dedupeStatuses; // Use provided version, or fall back to client's default version const effectiveVersion = version ?? this._defaultVersion; @@ -239,8 +240,8 @@ export class TaskHubGrpcClient { req.setVersion(v); } - if (orchestrationIdReusePolicy) { - req.setOrchestrationidreusepolicy(toProtobufOrchestrationIdReusePolicy(orchestrationIdReusePolicy)); + if (dedupeStatuses !== undefined) { + req.setOrchestrationidreusepolicy(toProtobufOrchestrationIdReusePolicy(dedupeStatuses)); } populateTagsMap(req.getTagsMap(), tags); @@ -263,6 +264,16 @@ 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 }); + } + } throw e; } finally { endSpan(span); @@ -584,7 +595,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 }); @@ -711,7 +726,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), @@ -826,7 +846,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); } @@ -979,7 +1002,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); } @@ -1242,10 +1269,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; } } diff --git a/packages/durabletask-js/src/index.ts b/packages/durabletask-js/src/index.ts index 06b83e80..bdf3e6ee 100644 --- a/packages/durabletask-js/src/index.ts +++ b/packages/durabletask-js/src/index.ts @@ -28,7 +28,8 @@ export { TERMINATE_OPTIONS_SYMBOL, } from "./orchestration/orchestration-terminate-options"; export { OrchestrationStatus } from "./orchestration/enum/orchestration-status.enum"; -export { OrchestrationIdReusePolicy } from "./orchestration/orchestration-id-reuse-policy"; +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/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 index ec95d78e..181e1890 100644 --- a/packages/durabletask-js/src/orchestration/orchestration-id-reuse-policy.ts +++ b/packages/durabletask-js/src/orchestration/orchestration-id-reuse-policy.ts @@ -4,40 +4,50 @@ import * as pb from "../proto/orchestrator_service_pb"; import { OrchestrationStatus, toProtobuf } from "./enum/orchestration-status.enum"; -const REPLACEABLE_RUNTIME_STATUSES: readonly OrchestrationStatus[] = [ - OrchestrationStatus.RUNNING, +/** Runtime statuses supported by orchestration ID deduplication, matching the .NET SDK. */ +export const ValidDedupeStatuses: readonly OrchestrationStatus[] = Object.freeze([ OrchestrationStatus.COMPLETED, OrchestrationStatus.FAILED, - OrchestrationStatus.CANCELED, OrchestrationStatus.TERMINATED, + OrchestrationStatus.CANCELED, + OrchestrationStatus.PENDING, + OrchestrationStatus.RUNNING, + OrchestrationStatus.SUSPENDED, +]); + +const RUNNING_STATUSES: readonly OrchestrationStatus[] = [ + OrchestrationStatus.RUNNING, OrchestrationStatus.PENDING, OrchestrationStatus.SUSPENDED, ]; -/** - * Controls whether a new orchestration rejects or atomically replaces an existing instance. - * - * The current shared protocol does not support an atomic no-op/IGNORE action. - */ -export interface OrchestrationIdReusePolicy { - /** - * Existing instances in one of these runtime statuses produce a duplicate-ID error. - * - * Instances in every other supported runtime status are atomically replaced. An empty - * list makes every supported runtime status replaceable. Omitting the policy preserves the - * backend's default duplicate-ID behavior. - */ - readonly dedupeStatuses: readonly OrchestrationStatus[]; +/** @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.`); + } + } + + const dedupeStatusSet = new Set(dedupeStatuses); + if ( + dedupeStatusSet.has(OrchestrationStatus.TERMINATED) && + RUNNING_STATUSES.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( - policy: OrchestrationIdReusePolicy, + dedupeStatuses: readonly OrchestrationStatus[], ): pb.OrchestrationIdReusePolicy { - const dedupeStatuses = new Set(policy.dedupeStatuses.map(toProtobuf)); - const replaceableStatuses = REPLACEABLE_RUNTIME_STATUSES.map(toProtobuf).filter( - (status) => !dedupeStatuses.has(status), - ); + 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); diff --git a/packages/durabletask-js/src/task/options/task-options.ts b/packages/durabletask-js/src/task/options/task-options.ts index 3838e1ea..1e3190e2 100644 --- a/packages/durabletask-js/src/task/options/task-options.ts +++ b/packages/durabletask-js/src/task/options/task-options.ts @@ -3,7 +3,7 @@ import { RetryPolicy } from "../retry/retry-policy"; import { AsyncRetryHandler, RetryHandler } from "../retry/retry-handler"; -import { OrchestrationIdReusePolicy } from "../../orchestration/orchestration-id-reuse-policy"; +import { OrchestrationStatus } from "../../orchestration/enum/orchestration-status.enum"; /** * Union type representing the available retry strategies for a task. @@ -76,10 +76,12 @@ export interface StartOrchestrationOptions { */ version?: string; /** - * Controls whether an existing orchestration with the same instance ID is - * deduplicated or atomically replaced based on its runtime status. + * Existing orchestration statuses that must produce a duplicate-ID error. + * + * An empty list makes every supported status replaceable. Omitting this property + * preserves the backend's default duplicate-ID behavior. */ - orchestrationIdReusePolicy?: OrchestrationIdReusePolicy; + 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 e52132bb..f2b1b18b 100644 --- a/packages/durabletask-js/src/testing/in-memory-backend.ts +++ b/packages/durabletask-js/src/testing/in-memory-backend.ts @@ -7,7 +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 { OrchestrationIdReusePolicy } from "../orchestration/orchestration-id-reuse-policy"; +import { validateDedupeStatuses } 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 { @@ -157,16 +158,9 @@ export class InMemoryOrchestrationBackend { input?: string, scheduledStartTime?: Date, parentInstance?: ParentOrchestrationInstance, - orchestrationIdReusePolicy?: OrchestrationIdReusePolicy, ): string { - const existingInstance = this.instances.get(instanceId); - if (existingInstance) { - const existingStatus = this.toClientStatus(existingInstance.status); - if (!orchestrationIdReusePolicy || orchestrationIdReusePolicy.dedupeStatuses.includes(existingStatus)) { - throw new Error(`Orchestration instance '${instanceId}' already exists`); - } - - this.removeInstanceForReplacement(instanceId); + if (this.instances.has(instanceId)) { + throw new OrchestrationAlreadyExistsError(`An orchestration with instance ID '${instanceId}' already exists`); } const now = new Date(); @@ -202,6 +196,66 @@ 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) { + validateDedupeStatuses(dedupeStatuses); + } + + const existingInstance = this.instances.get(instanceId); + if (existingInstance) { + const existingStatus = this.toClientStatus(existingInstance.status); + if (dedupeStatuses === undefined || dedupeStatuses.includes(existingStatus)) { + throw this.newAlreadyExistsError(instanceId, existingStatus); + } + + const existingExecutionId = existingInstance.executionId; + if (this.isRunningStatus(existingInstance.status)) { + const dedupeDescription = + 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)) { + throw this.newAlreadyExistsError(instanceId, currentStatus); + } + this.removeInstanceForReplacement(instanceId); + } + } + + return this.createInstance(instanceId, name, input, scheduledStartTime); + } + /** * Gets an orchestration instance by ID. */ @@ -639,9 +693,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 @@ -654,13 +706,7 @@ export class InMemoryOrchestrationBackend { /** * Completes an activity execution. */ - completeActivity( - instanceId: string, - executionId: 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 @@ -877,11 +923,7 @@ export class InMemoryOrchestrationBackend { } } - private cancelSubOrchestrationWatchers( - parentInstanceId: string, - parentExecutionId: string, - error: Error, - ): void { + private cancelSubOrchestrationWatchers(parentInstanceId: string, parentExecutionId: string, error: Error): void { for (const [instanceId, waiters] of this.stateWaiters) { const cancelledWaiters = waiters.filter( (waiter) => @@ -919,6 +961,28 @@ export class InMemoryOrchestrationBackend { 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 || @@ -1010,7 +1074,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); @@ -1160,11 +1230,7 @@ export class InMemoryOrchestrationBackend { }); } - 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; @@ -1624,10 +1690,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; } @@ -1651,10 +1714,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; } @@ -1664,9 +1724,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 18aea395..682a0480 100644 --- a/packages/durabletask-js/src/testing/test-client.ts +++ b/packages/durabletask-js/src/testing/test-client.ts @@ -62,21 +62,14 @@ export class TestOrchestrationClient { typeof instanceIdOrOptions === "string" || instanceIdOrOptions === undefined ? startAt : instanceIdOrOptions.startAt; - const orchestrationIdReusePolicy = + const dedupeStatuses = typeof instanceIdOrOptions === "string" || instanceIdOrOptions === undefined ? undefined - : instanceIdOrOptions.orchestrationIdReusePolicy; + : instanceIdOrOptions.dedupeStatuses; const id = instanceId ?? randomUUID(); const encodedInput = input !== undefined ? JSON.stringify(input) : undefined; - this.backend.createInstance( - id, - name, - encodedInput, - scheduledStartAt, - undefined, - orchestrationIdReusePolicy, - ); + await this.backend.createOrchestrationInstance(id, name, encodedInput, scheduledStartAt, dedupeStatuses); return id; } @@ -192,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)); } /** @@ -221,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, }; } diff --git a/packages/durabletask-js/test/orchestration-id-reuse-policy.spec.ts b/packages/durabletask-js/test/orchestration-id-reuse-policy.spec.ts index 2dd16db8..8e5a2ed5 100644 --- a/packages/durabletask-js/test/orchestration-id-reuse-policy.spec.ts +++ b/packages/durabletask-js/test/orchestration-id-reuse-policy.spec.ts @@ -6,12 +6,13 @@ import { ActivityContext, InMemoryOrchestrationBackend, OrchestrationContext, - OrchestrationIdReusePolicy, + 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"; @@ -30,6 +31,7 @@ function deferred(): { function mockStartInstance( client: TaskHubGrpcClient, captureRequest: (request: pb.CreateInstanceRequest) => void, + error?: grpc.ServiceError, ): void { const stub = (client as unknown as { _stub: Record })._stub; stub.startInstance = ( @@ -40,7 +42,7 @@ function mockStartInstance( captureRequest(request); const response = new pb.CreateInstanceResponse(); response.setInstanceid(request.getInstanceid()); - callback(null, response); + callback(error ?? null, response); return {} as grpc.ClientUnaryCall; }; } @@ -72,20 +74,16 @@ describe("TaskHubGrpcClient orchestration ID reuse policy", () => { mockStartInstance(client, (value) => { request = value; }); - const reusePolicy: OrchestrationIdReusePolicy = { - dedupeStatuses: [OrchestrationStatus.COMPLETED, OrchestrationStatus.FAILED], - }; - await client.scheduleNewOrchestration("workflow", undefined, { instanceId: "instance-1", - orchestrationIdReusePolicy: reusePolicy, + dedupeStatuses: [OrchestrationStatus.COMPLETED, OrchestrationStatus.FAILED], }); expect(request?.getOrchestrationidreusepolicy()?.getReplaceablestatusList()).toEqual([ - pb.OrchestrationStatus.ORCHESTRATION_STATUS_RUNNING, - pb.OrchestrationStatus.ORCHESTRATION_STATUS_CANCELED, 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, ]); }); @@ -98,19 +96,77 @@ describe("TaskHubGrpcClient orchestration ID reuse policy", () => { await client.scheduleNewOrchestration("workflow", undefined, { instanceId: "instance-1", - orchestrationIdReusePolicy: { dedupeStatuses: [] }, + dedupeStatuses: [], }); expect(request?.getOrchestrationidreusepolicy()?.getReplaceablestatusList()).toEqual([ - pb.OrchestrationStatus.ORCHESTRATION_STATUS_RUNNING, pb.OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED, pb.OrchestrationStatus.ORCHESTRATION_STATUS_FAILED, - pb.OrchestrationStatus.ORCHESTRATION_STATUS_CANCELED, 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("rejects terminated dedupe when a running status remains reusable", async () => { + let called = false; + mockStartInstance(client, () => { + called = true; + }); + + await expect( + client.scheduleNewOrchestration("workflow", undefined, { + 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.", + ), + ); + expect(called).toBe(false); + }); + + 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); + }); }); describe("TestOrchestrationClient orchestration ID reuse policy", () => { @@ -144,7 +200,7 @@ describe("TestOrchestrationClient orchestration ID reuse policy", () => { client.scheduleNewOrchestration(waitingOrchestrator, "replacement", { instanceId: "instance-1", }), - ).rejects.toThrow("already exists"); + ).rejects.toBeInstanceOf(OrchestrationAlreadyExistsError); }); it("rejects reuse when the existing runtime status is selected for deduplication", async () => { @@ -156,11 +212,9 @@ describe("TestOrchestrationClient orchestration ID reuse policy", () => { await expect( client.scheduleNewOrchestration(waitingOrchestrator, "replacement", { instanceId: "instance-1", - orchestrationIdReusePolicy: { - dedupeStatuses: [OrchestrationStatus.RUNNING], - }, + dedupeStatuses: [OrchestrationStatus.RUNNING], }), - ).rejects.toThrow("already exists"); + ).rejects.toBeInstanceOf(OrchestrationAlreadyExistsError); }); it("atomically replaces an existing instance whose runtime status is reusable", async () => { @@ -172,15 +226,12 @@ describe("TestOrchestrationClient orchestration ID reuse policy", () => { await client.scheduleNewOrchestration(waitingOrchestrator, "replacement", { instanceId: "instance-1", - orchestrationIdReusePolicy: { - dedupeStatuses: [ - OrchestrationStatus.COMPLETED, - OrchestrationStatus.FAILED, - OrchestrationStatus.TERMINATED, - OrchestrationStatus.PENDING, - OrchestrationStatus.SUSPENDED, - ], - }, + dedupeStatuses: [ + OrchestrationStatus.COMPLETED, + OrchestrationStatus.FAILED, + OrchestrationStatus.PENDING, + OrchestrationStatus.SUSPENDED, + ], }); const replacement = backend.getInstance("instance-1"); @@ -189,20 +240,21 @@ describe("TestOrchestrationClient orchestration ID reuse policy", () => { expect(replacement?.status).toBe(pb.OrchestrationStatus.ORCHESTRATION_STATUS_PENDING); }); - it("rejects waiters that were registered for the replaced execution", async () => { + 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 originalCompletionAssertion = expect(originalCompletion).rejects.toThrow("was replaced by a new execution"); - await client.scheduleNewOrchestration(waitingOrchestrator, "replacement", { + const replacement = client.scheduleNewOrchestration(waitingOrchestrator, "replacement", { instanceId: "instance-1", - orchestrationIdReusePolicy: { dedupeStatuses: [] }, + dedupeStatuses: [], }); - await originalCompletionAssertion; + const originalState = await originalCompletion; + expect(originalState?.runtimeStatus).toBe(OrchestrationStatus.TERMINATED); + await replacement; }); it("maps the canceled protobuf status to the public canceled status", () => { @@ -276,12 +328,20 @@ describe("TestOrchestrationClient replacement generation fences", () => { await client.scheduleNewOrchestration(orchestrator, "original", { instanceId: "activity-race" }); await originalActivityStarted.promise; - await client.scheduleNewOrchestration(orchestrator, "replacement", { + const replacement = client.scheduleNewOrchestration(orchestrator, "replacement", { instanceId: "activity-race", - orchestrationIdReusePolicy: { dedupeStatuses: [] }, + 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")); }); @@ -308,7 +368,7 @@ describe("TestOrchestrationClient replacement generation fences", () => { await client.scheduleNewOrchestration(parent, "replacement", { instanceId: "parent-race", - orchestrationIdReusePolicy: { dedupeStatuses: [] }, + dedupeStatuses: [], }); const replacementChildId = await replacementChildStarted.promise; @@ -347,13 +407,11 @@ describe("TestOrchestrationClient replacement generation fences", () => { await client.scheduleNewOrchestration(child, undefined, { instanceId: "replaced-child", - orchestrationIdReusePolicy: { dedupeStatuses: [] }, + dedupeStatuses: [], }); const state = await client.waitForOrchestrationCompletion("current-parent", true, 5); - expect(JSON.parse(state?.serializedOutput ?? "")).toContain( - "Sub-orchestration instance 'replaced-child' was replaced by a new execution", - ); + expect(JSON.parse(state?.serializedOutput ?? "")).toContain("Sub-orchestration failed"); }); it("removes child watchers owned by a replaced parent execution", async () => { @@ -383,7 +441,7 @@ describe("TestOrchestrationClient replacement generation fences", () => { await client.scheduleNewOrchestration(parent, "replacement", { instanceId: "replaced-parent", - orchestrationIdReusePolicy: { dedupeStatuses: [] }, + 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..1ef736d5 --- /dev/null +++ b/test/e2e-azuremanaged/orchestration-id-reuse-policy.spec.ts @@ -0,0 +1,94 @@ +// 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 endpoint = process.env.ENDPOINT || "localhost:8080"; +const taskHub = process.env.TASKHUB || "default"; + +describe("Orchestration ID reuse policy E2E", () => { + 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 = new DurableTaskAzureManagedClientBuilder().endpoint(endpoint, taskHub, null).build(); + worker = new DurableTaskAzureManagedWorkerBuilder().endpoint(endpoint, taskHub, null).build(); + 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("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); +}); From 2788bc5967ed295928b7fb2a57f87ec67f3c4eef Mon Sep 17 00:00:00 2001 From: wangbill Date: Thu, 6 Aug 2026 13:40:16 -0700 Subject: [PATCH 5/9] fix: align reuse validation boundaries Keep production gRPC validation wire-focused while applying shim-specific replacement checks and omitted-policy semantics only to the in-memory client. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: d491ec0d-9ce9-421b-9953-7d179d53625b --- CHANGELOG.md | 4 +- README.md | 8 ++- packages/durabletask-js/src/client/client.ts | 1 + .../orchestration-id-reuse-policy.ts | 14 ++-- .../src/task/options/task-options.ts | 5 +- .../src/testing/in-memory-backend.ts | 14 ++-- .../orchestration-id-reuse-policy.spec.ts | 69 +++++++++++++------ 7 files changed, 76 insertions(+), 39 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5f983eb7..46e44627 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,7 +5,9 @@ - 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 shared protocol does not support atomic no-op/`IGNORE`. + 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 diff --git a/README.md b/README.md index d9592b30..6064c123 100644 --- a/README.md +++ b/README.md @@ -99,9 +99,11 @@ await client.scheduleNewOrchestration(helloCities, undefined, { }); ``` -Omitting `dedupeStatuses` preserves the backend's default duplicate-ID behavior; passing `[]` -makes every supported runtime status replaceable. `ValidDedupeStatuses` exports the seven -supported statuses. The transient `CONTINUED_AS_NEW` status is not replaceable. A list containing +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 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. diff --git a/packages/durabletask-js/src/client/client.ts b/packages/durabletask-js/src/client/client.ts index 0d5f2072..9f0c3682 100644 --- a/packages/durabletask-js/src/client/client.ts +++ b/packages/durabletask-js/src/client/client.ts @@ -273,6 +273,7 @@ export class TaskHubGrpcClient { 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 { diff --git a/packages/durabletask-js/src/orchestration/orchestration-id-reuse-policy.ts b/packages/durabletask-js/src/orchestration/orchestration-id-reuse-policy.ts index 181e1890..25b39d0b 100644 --- a/packages/durabletask-js/src/orchestration/orchestration-id-reuse-policy.ts +++ b/packages/durabletask-js/src/orchestration/orchestration-id-reuse-policy.ts @@ -15,12 +15,6 @@ export const ValidDedupeStatuses: readonly OrchestrationStatus[] = Object.freeze OrchestrationStatus.SUSPENDED, ]); -const RUNNING_STATUSES: readonly OrchestrationStatus[] = [ - OrchestrationStatus.RUNNING, - OrchestrationStatus.PENDING, - OrchestrationStatus.SUSPENDED, -]; - /** @hidden Validates orchestration ID deduplication options. */ export function validateDedupeStatuses(dedupeStatuses: readonly OrchestrationStatus[]): void { for (const status of dedupeStatuses) { @@ -28,11 +22,17 @@ export function validateDedupeStatuses(dedupeStatuses: readonly OrchestrationSta 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) && - RUNNING_STATUSES.some((status) => !dedupeStatusSet.has(status)) + [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, " + diff --git a/packages/durabletask-js/src/task/options/task-options.ts b/packages/durabletask-js/src/task/options/task-options.ts index 1e3190e2..e876e8f2 100644 --- a/packages/durabletask-js/src/task/options/task-options.ts +++ b/packages/durabletask-js/src/task/options/task-options.ts @@ -78,8 +78,9 @@ export interface StartOrchestrationOptions { /** * Existing orchestration statuses that must produce a duplicate-ID error. * - * An empty list makes every supported status replaceable. Omitting this property - * preserves the backend's default duplicate-ID behavior. + * 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 f2b1b18b..a0d9a498 100644 --- a/packages/durabletask-js/src/testing/in-memory-backend.ts +++ b/packages/durabletask-js/src/testing/in-memory-backend.ts @@ -7,7 +7,7 @@ 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 { validateDedupeStatuses } from "../orchestration/orchestration-id-reuse-policy"; +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). */ @@ -207,20 +207,24 @@ export class InMemoryOrchestrationBackend { dedupeStatuses?: readonly ClientOrchestrationStatus[], ): Promise { if (dedupeStatuses !== undefined) { - validateDedupeStatuses(dedupeStatuses); + validateDedupeStatusesForReplacement(dedupeStatuses); } const existingInstance = this.instances.get(instanceId); if (existingInstance) { const existingStatus = this.toClientStatus(existingInstance.status); - if (dedupeStatuses === undefined || dedupeStatuses.includes(existingStatus)) { + if (dedupeStatuses?.includes(existingStatus) === true) { throw this.newAlreadyExistsError(instanceId, existingStatus); } const existingExecutionId = existingInstance.executionId; if (this.isRunningStatus(existingInstance.status)) { const dedupeDescription = - dedupeStatuses.length === 0 ? "[] (all statuses reusable)" : `[${dedupeStatuses.join(", ")}]`; + 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, ` + @@ -246,7 +250,7 @@ export class InMemoryOrchestrationBackend { const currentInstance = this.instances.get(instanceId); if (currentInstance?.executionId === existingExecutionId) { const currentStatus = this.toClientStatus(currentInstance.status); - if (dedupeStatuses.includes(currentStatus)) { + if (dedupeStatuses?.includes(currentStatus) === true) { throw this.newAlreadyExistsError(instanceId, currentStatus); } this.removeInstanceForReplacement(instanceId); diff --git a/packages/durabletask-js/test/orchestration-id-reuse-policy.spec.ts b/packages/durabletask-js/test/orchestration-id-reuse-policy.spec.ts index 8e5a2ed5..0da19524 100644 --- a/packages/durabletask-js/test/orchestration-id-reuse-policy.spec.ts +++ b/packages/durabletask-js/test/orchestration-id-reuse-policy.spec.ts @@ -136,23 +136,22 @@ describe("TaskHubGrpcClient orchestration ID reuse policy", () => { expect(called).toBe(false); }); - it("rejects terminated dedupe when a running status remains reusable", async () => { - let called = false; - mockStartInstance(client, () => { - called = true; + it("forwards terminated dedupe with a reusable running status to the sidecar", async () => { + let request: pb.CreateInstanceRequest | undefined; + mockStartInstance(client, (value) => { + request = value; }); - await expect( - client.scheduleNewOrchestration("workflow", undefined, { - 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.", - ), - ); - expect(called).toBe(false); + 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([ @@ -167,6 +166,16 @@ describe("TaskHubGrpcClient orchestration ID reuse policy", () => { 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", () => { @@ -191,16 +200,20 @@ describe("TestOrchestrationClient orchestration ID reuse policy", () => { backend.reset(); }); - it("preserves the default duplicate-ID error behavior", async () => { + 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 expect( - client.scheduleNewOrchestration(waitingOrchestrator, "replacement", { - instanceId: "instance-1", - }), - ).rejects.toBeInstanceOf(OrchestrationAlreadyExistsError); + 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 () => { @@ -217,6 +230,20 @@ describe("TestOrchestrationClient orchestration ID reuse policy", () => { ).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", From eb2ab621d4d128631ae738c08cc7da708ec5634f Mon Sep 17 00:00:00 2001 From: wangbill Date: Thu, 6 Aug 2026 13:53:58 -0700 Subject: [PATCH 6/9] test: run reuse policy E2E against Azure DTS Use the existing connection-string contract so the reuse-policy scenarios run against either authenticated Azure DTS or the local emulator. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: d491ec0d-9ce9-421b-9953-7d179d53625b --- .../orchestration-id-reuse-policy.spec.ts | 19 +++++++++++++++++-- 1 file changed, 17 insertions(+), 2 deletions(-) diff --git a/test/e2e-azuremanaged/orchestration-id-reuse-policy.spec.ts b/test/e2e-azuremanaged/orchestration-id-reuse-policy.spec.ts index 1ef736d5..8c3af25f 100644 --- a/test/e2e-azuremanaged/orchestration-id-reuse-policy.spec.ts +++ b/test/e2e-azuremanaged/orchestration-id-reuse-policy.spec.ts @@ -14,10 +14,25 @@ import { 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; @@ -32,8 +47,8 @@ describe("Orchestration ID reuse policy E2E", () => { }; beforeEach(async () => { - client = new DurableTaskAzureManagedClientBuilder().endpoint(endpoint, taskHub, null).build(); - worker = new DurableTaskAzureManagedWorkerBuilder().endpoint(endpoint, taskHub, null).build(); + client = createClient(); + worker = createWorker(); worker.addOrchestrator(reusableOrchestrator); await worker.start(); }); From aa12adf0f801e4a9809a786d540dabdc4add1e5f Mon Sep 17 00:00:00 2001 From: wangbill Date: Thu, 6 Aug 2026 14:00:55 -0700 Subject: [PATCH 7/9] test: document live DTS policy validation Cover the service-side rejection of self-conflicting status policies while preserving production request forwarding and the original suspended instance. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: d491ec0d-9ce9-421b-9953-7d179d53625b --- README.md | 9 +++--- .../orchestration-id-reuse-policy.spec.ts | 31 +++++++++++++++++++ 2 files changed, 36 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index 6064c123..aeaf9350 100644 --- a/README.md +++ b/README.md @@ -103,10 +103,11 @@ For `TaskHubGrpcClient`, omitting `dedupeStatuses` preserves the backend's defau 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 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. +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 diff --git a/test/e2e-azuremanaged/orchestration-id-reuse-policy.spec.ts b/test/e2e-azuremanaged/orchestration-id-reuse-policy.spec.ts index 8c3af25f..aa239008 100644 --- a/test/e2e-azuremanaged/orchestration-id-reuse-policy.spec.ts +++ b/test/e2e-azuremanaged/orchestration-id-reuse-policy.spec.ts @@ -93,6 +93,37 @@ describe("Orchestration ID reuse policy E2E", () => { 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 }); From 9e251485da06aaa70718686675d2e94415f653d0 Mon Sep 17 00:00:00 2001 From: wangbill Date: Thu, 6 Aug 2026 14:18:38 -0700 Subject: [PATCH 8/9] ci: run reuse policy DTS E2E Include the orchestration ID reuse policy spec in the existing orchestration emulator group for both supported Node versions. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: d491ec0d-9ce9-421b-9953-7d179d53625b --- .github/workflows/dts-e2e-tests.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/dts-e2e-tests.yaml b/.github/workflows/dts-e2e-tests.yaml index ede600b6..1823a6de 100644 --- a/.github/workflows/dts-e2e-tests.yaml +++ b/.github/workflows/dts-e2e-tests.yaml @@ -24,7 +24,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" From 40180606ac97245ae0704d09131a9540e91a8f0d Mon Sep 17 00:00:00 2001 From: wangbill Date: Thu, 6 Aug 2026 14:23:05 -0700 Subject: [PATCH 9/9] ci: enable manual DTS E2E runs Allow exact-head emulator verification without changing the existing push, pull request, or matrix coverage. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 6157e6cd-9ce9-4636-a8d5-4f49f4f32de8 --- .github/workflows/dts-e2e-tests.yaml | 1 + 1 file changed, 1 insertion(+) diff --git a/.github/workflows/dts-e2e-tests.yaml b/.github/workflows/dts-e2e-tests.yaml index 1823a6de..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