From 25fa0c400c1a0f1196b7fc14a0fa3e71b108b522 Mon Sep 17 00:00:00 2001 From: wangbill Date: Mon, 21 Sep 2026 12:51:12 -0700 Subject: [PATCH 1/2] feat: support versioned task registration and dispatch Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: e0af01a5-0dfa-4e71-a660-c4186e65d7e0 --- CHANGELOG.md | 9 + README.md | 78 +++++ packages/azure-functions-durable/CHANGELOG.md | 3 + packages/azure-functions-durable/README.md | 14 + .../test/unit/worker.spec.ts | 36 +++ .../durabletask-js-azuremanaged/CHANGELOG.md | 2 + .../durabletask-js-azuremanaged/README.md | 22 ++ .../src/worker-builder.ts | 40 ++- .../test/unit/worker-builder.spec.ts | 32 ++ .../src/task/options/task-options.ts | 8 +- .../src/testing/in-memory-backend.ts | 27 +- .../durabletask-js/src/testing/test-client.ts | 10 +- .../durabletask-js/src/testing/test-worker.ts | 40 ++- .../src/utils/pb-helper.util.ts | 5 + .../src/worker/activity-executor.ts | 5 +- .../activity-not-registered-error.ts | 4 +- .../orchestrator-not-registered-error.ts | 4 +- .../src/worker/orchestration-executor.ts | 15 +- .../durabletask-js/src/worker/registry.ts | 109 +++++-- .../worker/runtime-orchestration-context.ts | 17 +- .../src/worker/task-hub-grpc-worker.ts | 110 ++++--- .../src/worker/versioning-options.ts | 137 ++++---- .../src/worker/work-item-filters.ts | 29 +- .../orchestration-id-reuse-policy.spec.ts | 1 - .../test/versioned-dispatch-grpc.spec.ts | 239 ++++++++++++++ .../test/versioned-dispatch.spec.ts | 298 ++++++++++++++++++ .../test/versioned-in-memory.spec.ts | 160 ++++++++++ .../test/work-item-filters.spec.ts | 4 +- 28 files changed, 1243 insertions(+), 215 deletions(-) create mode 100644 packages/durabletask-js/test/versioned-dispatch-grpc.spec.ts create mode 100644 packages/durabletask-js/test/versioned-dispatch.spec.ts create mode 100644 packages/durabletask-js/test/versioned-in-memory.spec.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 894cc41b..80a550bc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,9 @@ ### New +- Add optional version arguments to orchestrator/activity registrations in core and test workers, + dispatch by recorded/request version, and preserve versions in the in-memory backend. Same-name + versions coexist; unversioned-only registrations retain the .NET-compatible fallback. - Align core and test worker timers with Python: three-day segments by default and `maximumTimerIntervalMs` override (`null`, zero, or negative disables segmentation). Positive fractions round up to milliseconds to match JavaScript Date precision. @@ -40,6 +43,12 @@ ### Breaking changes +- Activity calls without an explicit version now inherit the parent's instance version, and + child calls inherit the worker's `versioning.defaultVersion`; explicit `""` selects unversioned. + Worker acceptance checks now also apply to activities; `Strict` with an empty worker version + filters for unversioned work. Adding a versioned registration disables that name's unversioned + catch-all. Missing registrations are non-retriable. See README migration guidance before + changing defaults or upgrading workers with in-flight versioned instances. - Core timers now default to three-day segments instead of native timers; Azure-managed workers explicitly retain native timers. - `TimerTask.cancel()` now returns a boolean, marks the timer canceled and complete, and notifies diff --git a/README.md b/README.md index 078c0e22..e26572c1 100644 --- a/README.md +++ b/README.md @@ -274,6 +274,84 @@ const purchaseOrderWorkflow: TOrchestrator = async function* (ctx: Orchestration You can find the full sample at [examples/hello-world/human_interaction.ts](./examples/hello-world/human_interaction.ts). +### Versioned registration and dispatch + +One worker can host multiple implementations of the same orchestrator or activity name: + +```typescript +import { TaskHubGrpcClient, TaskHubGrpcWorker } from "@microsoft/durabletask-js"; + +const worker = new TaskHubGrpcWorker({ + versioning: { defaultVersion: "v2" }, + workItemFilters: "auto", +}); +worker.addNamedOrchestrator("Order", async function* (ctx) { + yield ctx.callActivity("Price"); +}, "v1"); +worker.addNamedOrchestrator("Order", async function* (ctx) { + yield ctx.callActivity("Price"); + yield ctx.callSubOrchestrator("Audit"); +}, "v2"); +worker.addNamedActivity("Price", () => 100, "v1"); +worker.addNamedActivity("Price", () => 120, "v2"); +worker.addNamedOrchestrator("Audit", () => "audited", "v2"); +await worker.start(); + +const client = new TaskHubGrpcClient(); +await client.scheduleNewOrchestration("Order", undefined, { version: "v1" }); +await client.scheduleNewOrchestration("Order", undefined, { version: "v2" }); +``` + +`addOrchestrator(fn, version?)` and `addActivity(fn, version?)` retain the function's native name; +the named variants accept `(name, fn, version?)`. These signatures also apply to +`TestOrchestrationWorker` and `DurableTaskAzureManagedWorkerBuilder`. Entities remain name-only. + +Versions are case-insensitive opaque strings: `"V1"` equals `"v1"`, but `"1"` and `"1.0"` are +different registrations. Omitted, `undefined`, `null` (JavaScript), and `""` versions identify the +unversioned registration. Whitespace-only registration versions are rejected; other strings are +not trimmed. Duplicate name/version pairs throw. **Task names remain case-sensitive in JavaScript**, +unlike .NET; this preserves existing JavaScript identifiers and function-name behavior. + +Dispatch uses the recorded `ExecutionStarted.version` during both initial execution and replay, +and `ActivityRequest.version` for activities. Exact matches win. Following the +[.NET factory contract](https://github.com/microsoft/durabletask-dotnet/blob/92474e9e35c66d64de36cabd0a17652376d37873/src/Worker/Core/DurableTaskFactory.cs), +a name with **only an unversioned registration** can handle any request version. Adding any +versioned registration for that name disables this fallback: unknown versions fail with the +existing not-registered error (including the requested version), marked non-retriable. An +unversioned request never selects a versioned implementation. + +| Scheduled work | When `options.version` is omitted | Explicit `version: ""` | +| --- | --- | --- | +| Activity | Current orchestration instance's `ctx.version` | Unversioned | +| Sub-orchestration | Worker's `versioning.defaultVersion`, or unversioned if unset | Unversioned | +| Top-level orchestration | Client's `defaultVersion`, or unversioned if unset | Unversioned | + +An explicit nonempty version overrides these defaults. Activities do **not** inherit the worker +default, and children do **not** inherit the parent's version. Both policy and handler retries +retain the originally selected version. Worker defaults do not change `continueAsNew` behavior. +The in-memory client accepts the start `version` option; its worker accepts +`{ versioning: { defaultVersion: "v2" } }`, but does not emulate gRPC worker acceptance/rejection. + +**Acceptance policy is separate from implementation lookup.** `Strict` and `CurrentOrOlder` +apply before dispatch to both orchestrations and activities; registrations cannot bypass them. +`Strict` with no worker version accepts only unversioned work. An omitted `matchStrategy` means +`None`. `Reject` abandons mismatched work; `Fail` returns an explicit non-retriable failure. +Auto filters contain one entry per logical name: all registered versions (including `""` for +mixed registrations), or a wildcard for unversioned-only names. Under `Strict`, filters use the +configured worker version, including `""`. Explicit filters are unchanged. + +**Migration:** previous JavaScript workers ignored activity request versions and always scheduled +unspecified activities as unversioned. Versioned parents now pass their version to activities; +use `{ version: "" }` to keep an activity unversioned. Setting a worker child default now affects +unspecified child calls; use explicit child versions for stable routing. Keep implementations for +all in-flight versions, and drain affected instances before changing replay-sensitive defaults or +mixing old and new workers. Adding the first versioned registration removes the name's legacy +catch-all, so register every version that still needs to run. + +Azure Functions `app.orchestration` / `app.activity` remain host registrations with unique function +names, not a multi-version host-routing API. The embedded `DurableFunctionsWorker` inherits core +version dispatch and child defaults for integrations that supply versioned protobuf requests. + ### Continue as new Long-running orchestrations can restart with fresh history and optionally move to a new diff --git a/packages/azure-functions-durable/CHANGELOG.md b/packages/azure-functions-durable/CHANGELOG.md index cffc386f..83b5e352 100644 --- a/packages/azure-functions-durable/CHANGELOG.md +++ b/packages/azure-functions-durable/CHANGELOG.md @@ -14,6 +14,9 @@ ### Fixes +- Inherit core version-aware replay dispatch and worker child defaults in the embedded + `DurableFunctionsWorker`, including classic-context wrappers. Host `app.*` function registrations + remain name-only; see README for the boundary and activity-version migration guidance. - Inherit Python-aligned core three-day timer/retry segments so the gRPC provider does not exceed Azure Storage's per-message delay limit. The testing helper inherits the same default. Functions also splits timers with DTS; there is no backend detection or Functions timer configuration. diff --git a/packages/azure-functions-durable/README.md b/packages/azure-functions-durable/README.md index a9d4296b..e4450768 100644 --- a/packages/azure-functions-durable/README.md +++ b/packages/azure-functions-durable/README.md @@ -20,6 +20,20 @@ This package supersedes the legacy [`durable-functions`](https://github.com/Azur - **Client** — `getClient(context)` returns a `DurableFunctionsClient` for scheduling, querying, signaling, and managing instances, plus HTTP management-payload helpers (`createCheckStatusResponse`, `createHttpManagementPayload`) for durable HTTP starters. The `app.client.*` starter helpers (`http`, `timer`, `storageBlob`, `storageQueue`, `serviceBusQueue`, `serviceBusTopic`, `eventHub`, `eventGrid`, `cosmosDB`, `generic`) register a normal trigger and inject the client as the handler's second argument, so `(trigger, client, context)` works without manually wiring `df.input.durableClient()` + `df.getClient(context)`. - **Classic (v3) compatibility** — orchestrators and entities written in the legacy `context.df.*` style, `RetryOptions`, `EntityId`, and the deprecated client aliases are adapted onto the core engine. +### Versioning boundary + +`app.orchestration` and `app.activity` register **unique Azure Function names**. They do not expose +multiple same-name versions to the Functions host; activities are dispatched by the host, not by +the core worker's activity registry. Existing app registrations remain unversioned catch-alls. + +For embedded integrations using `DurableFunctionsWorker` directly, the inherited core registration +methods accept an optional version and dispatch using the request's recorded version. Its constructor +accepts `versioning.defaultVersion` for child scheduling, including through the classic context +wrapper. Explicit child `version: ""` (the classic method's final argument) selects unversioned. +This low-level capability does not add a multi-version Functions host registration protocol. +See the core [versioning contract and migration guidance](../../README.md#versioned-registration-and-dispatch); +unspecified activities now carry their parent's instance version on the wire. + ## Migrating from durable-functions v3 This provider keeps classic `context.df.*` orchestrators and entities working, but a few surfaces diff --git a/packages/azure-functions-durable/test/unit/worker.spec.ts b/packages/azure-functions-durable/test/unit/worker.spec.ts index 38cd4583..e0c03b5d 100644 --- a/packages/azure-functions-durable/test/unit/worker.spec.ts +++ b/packages/azure-functions-durable/test/unit/worker.spec.ts @@ -30,6 +30,42 @@ async function timerActions(worker: DurableFunctionsWorker) { } describe("DurableFunctionsWorker", () => { + it("dispatches versioned registrations and child defaults through the classic wrapper", async () => { + const worker = new DurableFunctionsWorker({ + logger: new NoOpLogger(), + versioning: { defaultVersion: "child-default" }, + }); + for (const version of ["v1", "v2"]) { + worker.addNamedOrchestrator( + "Flow", + wrapOrchestrator(function* (ctx: ClassicOrchestrationContext) { + expect(ctx.df.version).toBe(version); + ctx.df.callSubOrchestrator("Child"); + ctx.df.callSubOrchestrator("Child", undefined, undefined, ""); + yield ctx.df.callSubOrchestrator("Child", undefined, undefined, "explicit"); + }), + version, + ); + } + for (const version of ["v1", "v2"]) { + const request = new pb.OrchestratorRequest() + .setInstanceid("instance") + .setNeweventsList([ + ph.newOrchestratorStartedEvent(START), + ph.newExecutionStartedEvent("Flow", "instance", undefined, undefined, undefined, version), + ]); + const response = await worker.handleOrchestratorRequest( + Buffer.from(request.serializeBinary()).toString("base64"), + ); + const actions = pb.OrchestratorResponse.deserializeBinary(Buffer.from(response, "base64")).getActionsList(); + expect(actions.map((action) => action.getCreatesuborchestration()?.getVersion()?.getValue() ?? "")).toEqual([ + "child-default", + "", + "explicit", + ]); + } + }); + it("inherits the core default without a Functions timer configuration surface", async () => { const options = { logger: new NoOpLogger(), maximumTimerIntervalMs: null }; const worker = new DurableFunctionsWorker(options); diff --git a/packages/durabletask-js-azuremanaged/CHANGELOG.md b/packages/durabletask-js-azuremanaged/CHANGELOG.md index b699b22d..1f763ee4 100644 --- a/packages/durabletask-js-azuremanaged/CHANGELOG.md +++ b/packages/durabletask-js-azuremanaged/CHANGELOG.md @@ -2,6 +2,8 @@ ### New +- Add an optional final version argument to all orchestrator/activity builder registrations, + preserving same-name versions and version-aware auto filters in the built worker. - Add `DurableTaskAzureManagedWorkerBuilder.silentDisconnectTimeout()` to configure the core worker's work-item stream watchdog. Set it to `0` for sidecars that do not emit health-ping work items, including the current durabletask-go sidecar. diff --git a/packages/durabletask-js-azuremanaged/README.md b/packages/durabletask-js-azuremanaged/README.md index 1f86b291..03453481 100644 --- a/packages/durabletask-js-azuremanaged/README.md +++ b/packages/durabletask-js-azuremanaged/README.md @@ -53,6 +53,28 @@ const worker = createAzureManagedWorkerBuilder("https://myservice.durabletask.io await worker.start(); ``` +### Versioned registrations + +All four orchestrator/activity registration methods accept an optional final `version` argument, +which is preserved when the builder constructs the worker: + +```typescript +const worker = createAzureManagedWorkerBuilder("http://localhost:8080", "myTaskHub", null) + .versioning({ defaultVersion: "v2" }) + .addNamedOrchestrator("Order", orderV1, "v1") + .addNamedOrchestrator("Order", orderV2, "v2") + .addNamedActivity("Price", priceV1, "v1") + .addNamedActivity("Price", priceV2, "v2") + .useWorkItemFilters() + .build(); +``` + +Names remain case-sensitive; versions are case-insensitive opaque strings. Omitted or empty +versions register an unversioned implementation. Worker filtering is independent of local +dispatch. The worker default applies to child orchestrations, while activities default to their +parent instance's version. See the core +[versioning and migration contract](../../README.md#versioned-registration-and-dispatch). + ## Supported Authentication Types The connection string `Authentication` parameter supports the following values: diff --git a/packages/durabletask-js-azuremanaged/src/worker-builder.ts b/packages/durabletask-js-azuremanaged/src/worker-builder.ts index 63db9c02..6b646bcb 100644 --- a/packages/durabletask-js-azuremanaged/src/worker-builder.ts +++ b/packages/durabletask-js-azuremanaged/src/worker-builder.ts @@ -24,8 +24,8 @@ import { export class DurableTaskAzureManagedWorkerBuilder { private _options: DurableTaskAzureManagedWorkerOptions; private _grpcChannelOptions: grpc.ChannelOptions = {}; - private _orchestrators: { name?: string; fn: TOrchestrator }[] = []; - private _activities: { name?: string; fn: TActivity }[] = []; + private _orchestrators: { name?: string; fn: TOrchestrator; version?: string }[] = []; + private _activities: { name?: string; fn: TActivity; version?: string }[] = []; private _entities: { name?: string; factory: EntityFactory }[] = []; private _logger: Logger = new ConsoleLogger(); private _shutdownTimeoutMs?: number; @@ -148,10 +148,11 @@ export class DurableTaskAzureManagedWorkerBuilder { * Registers an orchestrator function with the worker. * * @param fn The orchestrator function. + * @param version Registration version; omitted or empty means unversioned. * @returns This builder instance. */ - addOrchestrator(fn: TOrchestrator): DurableTaskAzureManagedWorkerBuilder { - this._orchestrators.push({ fn }); + addOrchestrator(fn: TOrchestrator, version?: string): DurableTaskAzureManagedWorkerBuilder { + this._orchestrators.push({ fn, version }); return this; } @@ -160,10 +161,11 @@ export class DurableTaskAzureManagedWorkerBuilder { * * @param name The name of the orchestrator. * @param fn The orchestrator function. + * @param version Registration version; omitted or empty means unversioned. * @returns This builder instance. */ - addNamedOrchestrator(name: string, fn: TOrchestrator): DurableTaskAzureManagedWorkerBuilder { - this._orchestrators.push({ name, fn }); + addNamedOrchestrator(name: string, fn: TOrchestrator, version?: string): DurableTaskAzureManagedWorkerBuilder { + this._orchestrators.push({ name, fn, version }); return this; } @@ -171,10 +173,11 @@ export class DurableTaskAzureManagedWorkerBuilder { * Registers an activity function with the worker. * * @param fn The activity function. + * @param version Registration version; omitted or empty means unversioned. * @returns This builder instance. */ - addActivity(fn: TActivity): DurableTaskAzureManagedWorkerBuilder { - this._activities.push({ fn }); + addActivity(fn: TActivity, version?: string): DurableTaskAzureManagedWorkerBuilder { + this._activities.push({ fn, version }); return this; } @@ -183,10 +186,15 @@ export class DurableTaskAzureManagedWorkerBuilder { * * @param name The name of the activity. * @param fn The activity function. + * @param version Registration version; omitted or empty means unversioned. * @returns This builder instance. */ - addNamedActivity(name: string, fn: TActivity): DurableTaskAzureManagedWorkerBuilder { - this._activities.push({ name, fn }); + addNamedActivity( + name: string, + fn: TActivity, + version?: string, + ): DurableTaskAzureManagedWorkerBuilder { + this._activities.push({ name, fn, version }); return this; } @@ -331,20 +339,20 @@ export class DurableTaskAzureManagedWorkerBuilder { }); // Register all orchestrators - for (const { name, fn } of this._orchestrators) { + for (const { name, fn, version } of this._orchestrators) { if (name) { - worker.addNamedOrchestrator(name, fn); + worker.addNamedOrchestrator(name, fn, version); } else { - worker.addOrchestrator(fn); + worker.addOrchestrator(fn, version); } } // Register all activities - for (const { name, fn } of this._activities) { + for (const { name, fn, version } of this._activities) { if (name) { - worker.addNamedActivity(name, fn); + worker.addNamedActivity(name, fn, version); } else { - worker.addActivity(fn); + worker.addActivity(fn, version); } } diff --git a/packages/durabletask-js-azuremanaged/test/unit/worker-builder.spec.ts b/packages/durabletask-js-azuremanaged/test/unit/worker-builder.spec.ts index cb849b29..848d85f6 100644 --- a/packages/durabletask-js-azuremanaged/test/unit/worker-builder.spec.ts +++ b/packages/durabletask-js-azuremanaged/test/unit/worker-builder.spec.ts @@ -23,6 +23,38 @@ describe("DurableTaskAzureManagedWorkerBuilder", () => { const ENDPOINT = "http://localhost:8080"; const TASKHUB = "test"; + it("preserves named and inferred version registrations and the child default on build", async () => { + const activity = () => "activity"; + const flow = async function* (ctx: import("@microsoft/durabletask-js").OrchestrationContext) { + yield ctx.callSubOrchestrator("Child"); + }; + const builder = new DurableTaskAzureManagedWorkerBuilder() + .endpoint(ENDPOINT, TASKHUB, null) + .versioning({ defaultVersion: "child" }); + for (const version of ["v1", "v2"]) { + builder.addOrchestrator(flow, version).addNamedOrchestrator("Named", flow, version); + builder.addActivity(activity, version).addNamedActivity("Named", activity, version); + } + const worker = builder.build(); + for (const name of ["flow", "Named"]) { + for (const version of ["v1", "v2"]) { + const request = new pb.OrchestratorRequest() + .setInstanceid("instance") + .setNeweventsList([ph.newExecutionStartedEvent(name, "instance", undefined, undefined, undefined, version)]); + const response = pb.OrchestratorResponse.deserializeBinary( + await worker.processOrchestratorRequest(request.serializeBinary()), + ); + expect(response.getActionsList()[0].getCreatesuborchestration()?.getVersion()?.getValue()).toBe("child"); + } + } + for (const name of ["activity", "Named"]) { + for (const version of ["v1", "v2"]) { + expect(worker["_registry"].getActivity(name, version)).toBe(activity); + } + } + expect(() => builder.addActivity(activity, "v1").build()).toThrow(/already exists/); + }); + it("keeps DTS timers native at their full deadline", async () => { const start = new Date("2026-01-01T00:00:00Z"); const deadline = new Date("2026-01-31T00:00:00Z"); diff --git a/packages/durabletask-js/src/task/options/task-options.ts b/packages/durabletask-js/src/task/options/task-options.ts index e876e8f2..691cebf0 100644 --- a/packages/durabletask-js/src/task/options/task-options.ts +++ b/packages/durabletask-js/src/task/options/task-options.ts @@ -34,7 +34,8 @@ export interface TaskOptions { tags?: Record; /** * The version of the task (activity) to execute. - * When specified, only workers that handle this version will process the task. + * Defaults to the current orchestration instance's version. An empty string explicitly + * selects an unversioned activity. Worker acceptance policy is checked before dispatch. */ version?: string; } @@ -44,6 +45,11 @@ export interface TaskOptions { * Extends TaskOptions with additional options specific to sub-orchestrations. */ export interface SubOrchestrationOptions extends TaskOptions { + /** + * The child version. Defaults to the worker's versioning.defaultVersion, not the parent's + * instance version. An empty string explicitly selects an unversioned child. + */ + version?: string; /** * The unique ID to use for the sub-orchestration instance. * If not specified, a deterministic ID will be generated based on the parent instance ID. diff --git a/packages/durabletask-js/src/testing/in-memory-backend.ts b/packages/durabletask-js/src/testing/in-memory-backend.ts index 6937381a..737c1226 100644 --- a/packages/durabletask-js/src/testing/in-memory-backend.ts +++ b/packages/durabletask-js/src/testing/in-memory-backend.ts @@ -43,6 +43,7 @@ export interface ActivityWorkItem { name: string; taskId: number; input?: string; + version?: string; completionToken: number; } @@ -158,6 +159,7 @@ export class InMemoryOrchestrationBackend { input?: string, scheduledStartTime?: Date, parentInstance?: ParentOrchestrationInstance, + version?: string, ): string { if (this.instances.has(instanceId)) { throw new OrchestrationAlreadyExistsError(`An orchestration with instance ID '${instanceId}' already exists`); @@ -185,7 +187,7 @@ export class InMemoryOrchestrationBackend { // Add initial events to start the orchestration const orchestratorStarted = pbh.newOrchestratorStartedEvent(startTime); - const executionStarted = pbh.newExecutionStartedEvent(name, instanceId, input, parentInstance, executionId); + const executionStarted = pbh.newExecutionStartedEvent(name, instanceId, input, parentInstance, executionId, version); instance.pendingEvents.push(orchestratorStarted); instance.pendingEvents.push(executionStarted); @@ -205,6 +207,7 @@ export class InMemoryOrchestrationBackend { input?: string, scheduledStartTime?: Date, dedupeStatuses?: readonly ClientOrchestrationStatus[], + version?: string, ): Promise { if (dedupeStatuses !== undefined) { validateDedupeStatusesForReplacement(dedupeStatuses); @@ -257,7 +260,7 @@ export class InMemoryOrchestrationBackend { } } - return this.createInstance(instanceId, name, input, scheduledStartTime); + return this.createInstance(instanceId, name, input, scheduledStartTime, undefined, version); } /** @@ -1108,6 +1111,7 @@ export class InMemoryOrchestrationBackend { // Add TaskScheduled event to history const event = pbh.newTaskScheduledEvent(taskId, taskName, input); + event.getTaskscheduled()!.setVersion(scheduleTask.getVersion()); instance.history.push(event); // Mark instance as running @@ -1120,6 +1124,7 @@ export class InMemoryOrchestrationBackend { instanceId: instance.instanceId, executionId: instance.executionId, name: taskName, + version: scheduleTask.getVersion()?.getValue(), taskId, input, completionToken: instance.completionToken, @@ -1173,6 +1178,7 @@ export class InMemoryOrchestrationBackend { // Add SubOrchestrationInstanceCreated event to history const event = pbh.newSubOrchestrationCreatedEvent(taskId, name, subInstanceId, input); + event.getSuborchestrationinstancecreated()!.setVersion(createSubOrch.getVersion()); instance.history.push(event); // Mark instance as running @@ -1182,11 +1188,18 @@ export class InMemoryOrchestrationBackend { // Create the sub-orchestration with parent instance info try { - this.createInstance(subInstanceId, name, input, undefined, { - name: instance.name, - instanceId: instance.instanceId, - taskScheduledId: taskId, - }); + this.createInstance( + subInstanceId, + name, + input, + undefined, + { + name: instance.name, + instanceId: instance.instanceId, + taskScheduledId: taskId, + }, + createSubOrch.getVersion()?.getValue(), + ); // Watch for sub-orchestration completion this.watchSubOrchestration(instance.instanceId, instance.executionId, subInstanceId, taskId); diff --git a/packages/durabletask-js/src/testing/test-client.ts b/packages/durabletask-js/src/testing/test-client.ts index 682a0480..cbb45d0b 100644 --- a/packages/durabletask-js/src/testing/test-client.ts +++ b/packages/durabletask-js/src/testing/test-client.ts @@ -25,8 +25,8 @@ 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. + * The in-memory backend does not model tags, so passing that option throws + * instead of silently diverging from TaskHubGrpcClient. */ async scheduleNewOrchestration( orchestrator: TOrchestrator | string, @@ -50,9 +50,6 @@ export class TestOrchestrationClient { 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 @@ -69,7 +66,8 @@ export class TestOrchestrationClient { const id = instanceId ?? randomUUID(); const encodedInput = input !== undefined ? JSON.stringify(input) : undefined; - await this.backend.createOrchestrationInstance(id, name, encodedInput, scheduledStartAt, dedupeStatuses); + const version = typeof instanceIdOrOptions === "object" ? instanceIdOrOptions.version : undefined; + await this.backend.createOrchestrationInstance(id, name, encodedInput, scheduledStartAt, dedupeStatuses, version); return id; } diff --git a/packages/durabletask-js/src/testing/test-worker.ts b/packages/durabletask-js/src/testing/test-worker.ts index 17c45ecd..38abbff2 100644 --- a/packages/durabletask-js/src/testing/test-worker.ts +++ b/packages/durabletask-js/src/testing/test-worker.ts @@ -22,6 +22,7 @@ import * as pb from "../proto/orchestrator_service_pb"; import * as pbh from "../utils/pb-helper.util"; import type { TaskHubGrpcWorkerOptions } from "../worker/task-hub-grpc-worker"; import { resolveMaximumTimerInterval } from "../worker/timer-interval"; +import type { VersioningOptions } from "../worker/versioning-options"; /** * Worker that processes orchestrations and activities from the in-memory backend. @@ -29,6 +30,8 @@ import { resolveMaximumTimerInterval } from "../worker/timer-interval"; * This worker runs in the same process as the test and processes work items * synchronously in the Node.js event loop, avoiding the need for a separate * sidecar process. + * Versioned registrations and child defaults are supported; backend work-item filtering + * and the gRPC worker's version acceptance/rejection policy are not emulated. */ export class TestOrchestrationWorker { private readonly registry: Registry; @@ -37,55 +40,63 @@ export class TestOrchestrationWorker { private processingPromise: Promise | null = null; private stopRequested: boolean = false; private readonly maximumTimerIntervalMs: number | null; + private readonly defaultVersion?: string; constructor( backend: InMemoryOrchestrationBackend, - options: Pick = {}, + options: Pick & { + versioning?: Pick; + } = {}, ) { this.maximumTimerIntervalMs = resolveMaximumTimerInterval(options.maximumTimerIntervalMs); + this.defaultVersion = options.versioning?.defaultVersion; this.registry = new Registry(); this.backend = backend; } /** * Registers an orchestrator function with the worker. + * Omit version (or use "") for an unversioned registration. */ - addOrchestrator(fn: TOrchestrator): string { + addOrchestrator(fn: TOrchestrator, version?: string): string { if (this.isRunning) { throw new Error("Cannot add orchestrator while worker is running."); } - return this.registry.addOrchestrator(fn); + return this.registry.addOrchestrator(fn, version); } /** * Registers a named orchestrator function with the worker. + * Omit version (or use "") for an unversioned registration. */ - addNamedOrchestrator(name: string, fn: TOrchestrator): string { + addNamedOrchestrator(name: string, fn: TOrchestrator, version?: string): string { if (this.isRunning) { throw new Error("Cannot add orchestrator while worker is running."); } - this.registry.addNamedOrchestrator(name, fn); + this.registry.addNamedOrchestrator(name, fn, version); return name; } /** * Registers an activity function with the worker. + * Omit version (or use "") for an unversioned registration. */ - addActivity(fn: TActivity): string { + addActivity(fn: TActivity, version?: string): string { if (this.isRunning) { throw new Error("Cannot add activity while worker is running."); } - return this.registry.addActivity(fn); + return this.registry.addActivity(fn, version); } /** * Registers a named activity function with the worker. + * Omit version (or use "") for an unversioned registration. */ - addNamedActivity(name: string, fn: TActivity): string { + addNamedActivity(name: string, fn: TActivity, version?: string): string { if (this.isRunning) { throw new Error("Cannot add activity while worker is running."); } - this.registry.addNamedActivity(name, fn); + this.registry.addNamedActivity(name, fn, version); return name; } @@ -185,7 +196,12 @@ export class TestOrchestrationWorker { const completionToken = instance.completionToken; try { - const executor = new OrchestrationExecutor(this.registry, undefined, this.maximumTimerIntervalMs); + const executor = new OrchestrationExecutor( + this.registry, + undefined, + this.maximumTimerIntervalMs, + this.defaultVersion, + ); const result = await executor.execute(instanceId, instance.history, instance.pendingEvents, instance.executionId); this.backend.completeOrchestration(instanceId, completionToken, result.actions, result.customStatus); @@ -207,11 +223,11 @@ export class TestOrchestrationWorker { * Processes a single activity work item. */ private async processActivity(workItem: ActivityWorkItem): Promise { - const { instanceId, executionId, name, taskId, input } = workItem; + const { instanceId, executionId, name, taskId, input, version } = workItem; try { const executor = new ActivityExecutor(this.registry); - const result = await executor.execute(instanceId, name, taskId, input); + const result = await executor.execute(instanceId, name, taskId, input, version); this.backend.completeActivity(instanceId, executionId, taskId, result); } catch (error: unknown) { const err = error instanceof Error ? error : new Error(String(error)); diff --git a/packages/durabletask-js/src/utils/pb-helper.util.ts b/packages/durabletask-js/src/utils/pb-helper.util.ts index 45e500cb..8cb61000 100644 --- a/packages/durabletask-js/src/utils/pb-helper.util.ts +++ b/packages/durabletask-js/src/utils/pb-helper.util.ts @@ -4,6 +4,8 @@ import { StringValue } from "google-protobuf/google/protobuf/wrappers_pb"; import * as pb from "../proto/orchestrator_service_pb"; import { Timestamp } from "google-protobuf/google/protobuf/timestamp_pb"; +import { ActivityNotRegisteredError } from "../worker/exception/activity-not-registered-error"; +import { OrchestratorNotRegisteredError } from "../worker/exception/orchestrator-not-registered-error"; export function newOrchestratorStartedEvent(timestamp?: Date | null): pb.HistoryEvent { const ts = new Timestamp(); @@ -220,6 +222,9 @@ function buildFailureDetails(e: unknown, depth: number): pb.TaskFailureDetails { failure.setErrortype(errorType); failure.setErrormessage(errorMessage); + if (e instanceof ActivityNotRegisteredError || e instanceof OrchestratorNotRegisteredError) { + failure.setIsnonretriable(true); + } if (stack !== undefined) { const sv = new StringValue(); diff --git a/packages/durabletask-js/src/worker/activity-executor.ts b/packages/durabletask-js/src/worker/activity-executor.ts index f4ad151a..65efbe21 100644 --- a/packages/durabletask-js/src/worker/activity-executor.ts +++ b/packages/durabletask-js/src/worker/activity-executor.ts @@ -22,11 +22,12 @@ export class ActivityExecutor { name: string, taskId: number, encodedInput?: string, + version?: string, ): Promise { - const fn = this._registry.getActivity(name); + const fn = this._registry.getActivity(name, version); if (!fn) { - throw new ActivityNotRegisteredError(name); + throw new ActivityNotRegisteredError(name, version); } // Log activity start (EventId 603) diff --git a/packages/durabletask-js/src/worker/exception/activity-not-registered-error.ts b/packages/durabletask-js/src/worker/exception/activity-not-registered-error.ts index a4df13b7..30a732e4 100644 --- a/packages/durabletask-js/src/worker/exception/activity-not-registered-error.ts +++ b/packages/durabletask-js/src/worker/exception/activity-not-registered-error.ts @@ -2,8 +2,8 @@ // Licensed under the MIT License. export class ActivityNotRegisteredError extends Error { - constructor(name: string) { - super(`Activity function '${name}' is not registered.`); + constructor(name: string, version?: string) { + super(`Activity function '${name}'${version ? ` with version '${version}'` : ""} is not registered.`); this.name = "ActivityNotRegisteredError"; } } diff --git a/packages/durabletask-js/src/worker/exception/orchestrator-not-registered-error.ts b/packages/durabletask-js/src/worker/exception/orchestrator-not-registered-error.ts index d8b6ae08..ec5d0ed7 100644 --- a/packages/durabletask-js/src/worker/exception/orchestrator-not-registered-error.ts +++ b/packages/durabletask-js/src/worker/exception/orchestrator-not-registered-error.ts @@ -2,8 +2,8 @@ // Licensed under the MIT License. export class OrchestratorNotRegisteredError extends Error { - constructor(name?: string) { - super(`Orchestrator '${name}' does not exist.`); + constructor(name?: string, version?: string) { + super(`Orchestrator '${name}'${version ? ` with version '${version}'` : ""} does not exist.`); this.name = "OrchestratorNotRegisteredError"; } } diff --git a/packages/durabletask-js/src/worker/orchestration-executor.ts b/packages/durabletask-js/src/worker/orchestration-executor.ts index 8f02b678..04392ce3 100644 --- a/packages/durabletask-js/src/worker/orchestration-executor.ts +++ b/packages/durabletask-js/src/worker/orchestration-executor.ts @@ -50,7 +50,12 @@ export class OrchestrationExecutor { private _orchestratorName: string; private readonly _maximumTimerIntervalMs: number | null; - constructor(registry: Registry, logger?: Logger, maximumTimerIntervalMs?: number | null) { + constructor( + registry: Registry, + logger?: Logger, + maximumTimerIntervalMs?: number | null, + private readonly _defaultVersion?: string, + ) { this._maximumTimerIntervalMs = resolveMaximumTimerInterval(maximumTimerIntervalMs); this._registry = registry; this._generator = undefined; @@ -87,7 +92,7 @@ export class OrchestrationExecutor { return buildRewindResult(oldEvents, newEvents); } - const ctx = new RuntimeOrchestrationContext(instanceId, this._maximumTimerIntervalMs); + const ctx = new RuntimeOrchestrationContext(instanceId, this._maximumTimerIntervalMs, this._defaultVersion); // Seed the execution ID from the authoritative source (the OrchestratorRequest on the gRPC path, // or the backend record on the in-memory path). The ExecutionStarted event replayed below may // also carry it; handleExecutionStarted reconciles the two. @@ -255,10 +260,14 @@ export class OrchestrationExecutor { const executionStartedEvent = event.getExecutionstarted(); const fn = this._registry.getOrchestrator( executionStartedEvent ? executionStartedEvent.getName() : undefined, + executionStartedEvent?.getVersion()?.getValue(), ); if (!fn) { - throw new OrchestratorNotRegisteredError(executionStartedEvent?.getName()); + throw new OrchestratorNotRegisteredError( + executionStartedEvent?.getName(), + executionStartedEvent?.getVersion()?.getValue(), + ); } // Set the execution ID from the orchestration instance. If the executor was already seeded with diff --git a/packages/durabletask-js/src/worker/registry.ts b/packages/durabletask-js/src/worker/registry.ts index 9fc5dc3a..0389cc53 100644 --- a/packages/durabletask-js/src/worker/registry.ts +++ b/packages/durabletask-js/src/worker/registry.ts @@ -7,78 +7,107 @@ import { TOrchestrator } from "../types/orchestrator.type"; import { TOutput } from "../types/output.type"; import { EntityFactory } from "../entities/task-entity"; +interface TaskRegistration { + readonly name: string; + readonly version: string; + readonly fn: T; +} + +type TaskRegistrations = Map>>; + +function addRegistration( + registrations: TaskRegistrations, + kind: string, + name: string, + fn: T, + version = "", +): void { + if (!name) { + throw new Error(`A non-empty ${kind} name is required.`); + } + version = version ?? ""; + if (version && !version.trim()) { + throw new Error("A task version must not contain only whitespace."); + } + const key = version.toLowerCase(); + const versions = registrations.get(name) ?? new Map>(); + if (versions.has(key)) { + const versionText = version ? ` with version '${version}'` : ""; + throw new Error(`A '${name}' ${kind}${versionText} already exists.`); + } + versions.set(key, { name, version, fn }); + registrations.set(name, versions); +} + +function getRegisteredTask(registrations: TaskRegistrations, name: string, version?: string): T | undefined { + const versions = registrations.get(name); + // Like the .NET processor, treat whitespace-only wire versions as unversioned. + const key = version?.trim() ? version.toLowerCase() : ""; + const exact = versions?.get(key); + if (exact) { + return exact.fn; + } + // Legacy registrations handle any version only until this name opts into versioned dispatch. + return key && versions?.size === 1 ? versions.get("")?.fn : undefined; +} + /** * Registry for orchestrators, activities, and entities. * * @remarks - * This class is used by the worker to look up task implementations by name. + * Task names are case-sensitive; versions are case-insensitive opaque strings. + * An omitted or empty version registers the unversioned implementation. * Entity names are normalized to lowercase for case-insensitive matching. */ export class Registry { - private _orchestrators: Record; - private _activities: Record>; + private _orchestrators: TaskRegistrations; + private _activities: TaskRegistrations>; private _entities: Record; constructor() { - this._orchestrators = {}; - this._activities = {}; + this._orchestrators = new Map(); + this._activities = new Map(); this._entities = {}; } - addOrchestrator(fn: TOrchestrator): string { + addOrchestrator(fn: TOrchestrator, version?: string): string { if (!fn) { throw new Error("An orchestrator function argument is required."); } const name = this._getFunctionName(fn); - this.addNamedOrchestrator(name, fn); + this.addNamedOrchestrator(name, fn, version); return name; } - addNamedOrchestrator(name: string, fn: TOrchestrator): void { - if (!name) { - throw new Error("A non-empty orchestrator name is required."); - } - - if (name in this._orchestrators) { - throw new Error(`A '${name}' orchestrator already exists.`); - } - - this._orchestrators[name] = fn; + addNamedOrchestrator(name: string, fn: TOrchestrator, version?: string): void { + addRegistration(this._orchestrators, "orchestrator", name, fn, version); } - getOrchestrator(name?: string): TOrchestrator | undefined { + getOrchestrator(name?: string, version?: string): TOrchestrator | undefined { if (!name) { return undefined; } - return this._orchestrators[name]; + return getRegisteredTask(this._orchestrators, name, version); } - addActivity(fn: TActivity): string { + addActivity(fn: TActivity, version?: string): string { if (!fn) { throw new Error("An activity function argument is required."); } const name = this._getFunctionName(fn); - this.addNamedActivity(name, fn); + this.addNamedActivity(name, fn, version); return name; } - addNamedActivity(name: string, fn: TActivity): void { - if (!name) { - throw new Error("A non-empty activity name is required."); - } - - if (name in this._activities) { - throw new Error(`A '${name}' activity already exists.`); - } - - this._activities[name] = fn; + addNamedActivity(name: string, fn: TActivity, version?: string): void { + addRegistration(this._activities, "activity", name, fn, version); } - getActivity(name: string): TActivity | undefined { - return this._activities[name]; + getActivity(name: string, version?: string): TActivity | undefined { + return getRegisteredTask(this._activities, name, version); } /** @@ -152,14 +181,24 @@ export class Registry { * Gets the names of all registered orchestrators. */ getOrchestratorNames(): string[] { - return Object.keys(this._orchestrators); + return [...this._orchestrators.keys()]; } /** * Gets the names of all registered activities. */ getActivityNames(): string[] { - return Object.keys(this._activities); + return [...this._activities.keys()]; + } + + /** Gets every orchestrator registration, including versions sharing the same name. */ + getOrchestratorRegistrations(): TaskRegistration[] { + return [...this._orchestrators.values()].flatMap((versions) => [...versions.values()]); + } + + /** Gets every activity registration, including versions sharing the same name. */ + getActivityRegistrations(): TaskRegistration>[] { + return [...this._activities.values()].flatMap((versions) => [...versions.values()]); } /** diff --git a/packages/durabletask-js/src/worker/runtime-orchestration-context.ts b/packages/durabletask-js/src/worker/runtime-orchestration-context.ts index 76b2a100..4c45e0cc 100644 --- a/packages/durabletask-js/src/worker/runtime-orchestration-context.ts +++ b/packages/durabletask-js/src/worker/runtime-orchestration-context.ts @@ -57,7 +57,11 @@ export class RuntimeOrchestrationContext extends OrchestrationContext { private readonly _maximumTimerIntervalMs: number | null; private readonly _longTimers = new Map, number>(); - constructor(instanceId: string, maximumTimerIntervalMs?: number | null) { + constructor( + instanceId: string, + maximumTimerIntervalMs?: number | null, + private readonly _defaultVersion?: string, + ) { super(); this._maximumTimerIntervalMs = resolveMaximumTimerInterval(maximumTimerIntervalMs); @@ -406,7 +410,7 @@ export class RuntimeOrchestrationContext extends OrchestrationContext { const id = this.nextSequenceNumber(); const name = typeof activity === "string" ? activity : getName(activity); const encodedInput = input !== undefined ? JSON.stringify(input) : undefined; - const action = ph.newScheduleTaskAction(id, name, encodedInput, options?.tags, options?.version); + const action = ph.newScheduleTaskAction(id, name, encodedInput, options?.tags, options?.version ?? this._version); this._pendingActions[action.getId()] = action; const task = this.createRetryTaskOrDefault(action, id, options, "activity"); @@ -460,7 +464,14 @@ export class RuntimeOrchestrationContext extends OrchestrationContext { } const encodedInput = input !== undefined ? JSON.stringify(input) : undefined; - const action = ph.newCreateSubOrchestrationAction(id, name, instanceId, encodedInput, options?.tags, options?.version); + const action = ph.newCreateSubOrchestrationAction( + id, + name, + instanceId, + encodedInput, + options?.tags, + options?.version ?? this._defaultVersion, + ); this._pendingActions[action.getId()] = action; const task = this.createRetryTaskOrDefault(action, id, options, "subOrchestration"); diff --git a/packages/durabletask-js/src/worker/task-hub-grpc-worker.ts b/packages/durabletask-js/src/worker/task-hub-grpc-worker.ts index 7b0d656e..a52d0481 100644 --- a/packages/durabletask-js/src/worker/task-hub-grpc-worker.ts +++ b/packages/durabletask-js/src/worker/task-hub-grpc-worker.ts @@ -95,7 +95,7 @@ export interface TaskHubGrpcWorkerOptions { * Defaults to 5. Non-positive values disable channel recreation. Must be a safe integer. */ channelRecreateFailureThreshold?: number; - /** Optional versioning options for filtering orchestrations by version. */ + /** Version acceptance policy for orchestrations and activities, and the default child version. */ versioning?: VersioningOptions; /** * Optional work item filters to control which work items the worker receives. @@ -274,59 +274,63 @@ export class TaskHubGrpcWorker { /** * Registers an orchestrator function with the worker. + * An optional version permits same-name implementations. Omission or "" is unversioned. * * @param fn * @returns */ - addOrchestrator(fn: TOrchestrator): string { + addOrchestrator(fn: TOrchestrator, version?: string): string { if (this._isRunning) { throw new Error("Cannot add orchestrator while worker is running."); } - return this._registry.addOrchestrator(fn); + return this._registry.addOrchestrator(fn, version); } /** * Registers an named orchestrator function with the worker. + * An optional version permits same-name implementations. Omission or "" is unversioned. * * @param fn * @returns */ - addNamedOrchestrator(name: string, fn: TOrchestrator): string { + addNamedOrchestrator(name: string, fn: TOrchestrator, version?: string): string { if (this._isRunning) { throw new Error("Cannot add orchestrator while worker is running."); } - this._registry.addNamedOrchestrator(name, fn); + this._registry.addNamedOrchestrator(name, fn, version); return name; } /** * Registers an activity function with the worker. + * An optional version permits same-name implementations. Omission or "" is unversioned. * * @param fn * @returns */ - addActivity(fn: TActivity): string { + addActivity(fn: TActivity, version?: string): string { if (this._isRunning) { throw new Error("Cannot add activity while worker is running."); } - return this._registry.addActivity(fn); + return this._registry.addActivity(fn, version); } /** * Registers an named activity function with the worker. + * An optional version permits same-name implementations. Omission or "" is unversioned. * * @param fn * @returns */ - addNamedActivity(name: string, fn: TActivity): string { + addNamedActivity(name: string, fn: TActivity, version?: string): string { if (this._isRunning) { throw new Error("Cannot add activity while worker is running."); } - this._registry.addNamedActivity(name, fn); + this._registry.addNamedActivity(name, fn, version); return name; } @@ -836,7 +840,7 @@ export class TaskHubGrpcWorker { /** * Result of version compatibility check. */ - private _checkVersionCompatibility(req: pb.OrchestratorRequest): { + private _checkVersionCompatibility(orchestrationVersion?: string): { compatible: boolean; shouldFail: boolean; orchestrationVersion?: string; @@ -844,18 +848,14 @@ export class TaskHubGrpcWorker { errorMessage?: string; } { // If no versioning options configured or match strategy is None, always compatible - if (!this._versioning || this._versioning.matchStrategy === VersionMatchStrategy.None) { + if ( + !this._versioning || + (this._versioning.matchStrategy ?? VersionMatchStrategy.None) === VersionMatchStrategy.None + ) { return { compatible: true, shouldFail: false }; } - // Extract orchestration version from ExecutionStarted event - const orchestrationVersion = this._getOrchestrationVersion(req); - const workerVersion = this._versioning.version; - - // If worker version is not set, process all - if (!workerVersion) { - return { compatible: true, shouldFail: false }; - } + const workerVersion = this._versioning.version ?? ""; let compatible = false; let errorType = "VersionMismatch"; @@ -1079,7 +1079,7 @@ export class TaskHubGrpcWorker { } // Check version compatibility if versioning is enabled - const versionCheckResult = this._checkVersionCompatibility(req); + const versionCheckResult = this._checkVersionCompatibility(this._getOrchestrationVersion(req)); if (!versionCheckResult.compatible) { if (versionCheckResult.shouldFail) { // Fail the orchestration with version mismatch error @@ -1170,7 +1170,12 @@ export class TaskHubGrpcWorker { let res; try { - const executor = new OrchestrationExecutor(this._registry, this._logger, this.maximumTimerIntervalMs); + const executor = new OrchestrationExecutor( + this._registry, + this._logger, + this.maximumTimerIntervalMs, + this._versioning?.defaultVersion, + ); const result = await executor.execute( req.getInstanceid(), req.getPasteventsList(), @@ -1272,30 +1277,51 @@ export class TaskHubGrpcWorker { throw new Error("Activity request does not contain an orchestration instance id"); } - let res; - - // Start the activity span for distributed tracing - const activitySpan = startSpanForTaskExecution(req); - - try { - const executor = new ActivityExecutor(this._registry, this._logger); - const result = await executor.execute( + const versionCheck = this._checkVersionCompatibility(req.getVersion()?.getValue()); + if (!versionCheck.compatible && !versionCheck.shouldFail) { + WorkerLogs.versionMismatchAbandon( + this._logger, instanceId, - req.getName(), - req.getTaskid(), - req.getInput()?.getValue() ?? "", + versionCheck.errorType!, + versionCheck.errorMessage!, ); + try { + const abandon = new pb.AbandonActivityTaskRequest().setCompletiontoken(completionToken); + await this._deliverResponse(stub.abandonTaskActivityWorkItem.bind(stub), abandon, retrySignal); + } catch (error: unknown) { + WorkerLogs.completionError(this._logger, instanceId, error); + } + return; + } - const s = new StringValue(); - s.setValue(result ?? ""); + const res = new pb.ActivityResponse() + .setInstanceid(instanceId) + .setTaskid(req.getTaskid()) + .setCompletiontoken(completionToken); - res = new pb.ActivityResponse(); - res.setInstanceid(instanceId); - res.setTaskid(req.getTaskid()); - res.setCompletiontoken(completionToken); - res.setResult(s); + // Start the activity span for distributed tracing + const activitySpan = startSpanForTaskExecution(req); - setSpanOk(activitySpan); + try { + if (!versionCheck.compatible) { + const error = new Error(versionCheck.errorMessage); + WorkerLogs.activityExecutionError(this._logger, req.getName(), error); + setSpanError(activitySpan, error); + res.setFailuredetails( + pbh.newVersionMismatchFailureDetails(versionCheck.errorType!, versionCheck.errorMessage!), + ); + } else { + const executor = new ActivityExecutor(this._registry, this._logger); + const result = await executor.execute( + instanceId, + req.getName(), + req.getTaskid(), + req.getInput()?.getValue() ?? "", + req.getVersion()?.getValue(), + ); + res.setResult(new StringValue().setValue(result ?? "")); + setSpanOk(activitySpan); + } } catch (e: unknown) { const error = e instanceof Error ? e : new Error(String(e)); WorkerLogs.activityExecutionError(this._logger, req.getName(), error); @@ -1304,10 +1330,6 @@ export class TaskHubGrpcWorker { const failureDetails = pbh.newFailureDetails(error); - res = new pb.ActivityResponse(); - res.setInstanceid(instanceId); - res.setTaskid(req.getTaskid()); - res.setCompletiontoken(completionToken); res.setFailuredetails(failureDetails); } finally { // End the activity span BEFORE the gRPC completion call. diff --git a/packages/durabletask-js/src/worker/versioning-options.ts b/packages/durabletask-js/src/worker/versioning-options.ts index 81713126..991f7bea 100644 --- a/packages/durabletask-js/src/worker/versioning-options.ts +++ b/packages/durabletask-js/src/worker/versioning-options.ts @@ -1,68 +1,69 @@ -// Copyright (c) Microsoft Corporation. -// Licensed under the MIT License. - -/** - * Strategy for matching orchestration versions when processing work items. - */ -export enum VersionMatchStrategy { - /** - * No version matching - process all orchestrations regardless of version. - */ - None = 0, - - /** - * Only process orchestrations that exactly match the worker's version. - */ - Strict = 1, - - /** - * Process orchestrations with the current version or older versions. - * Uses semantic versioning comparison. - */ - CurrentOrOlder = 2, -} - -/** - * Strategy for handling version mismatches when processing work items. - */ -export enum VersionFailureStrategy { - /** - * Reject the work item and let it be picked up by another worker. - * The orchestration will be retried by a compatible worker. - */ - Reject = 0, - - /** - * Fail the orchestration with a version mismatch error. - * This will mark the orchestration as failed. - */ - Fail = 1, -} - -/** - * Options for configuring version-based filtering of orchestrations. - */ -export interface VersioningOptions { - /** - * The version of the worker. This is used for version matching when processing orchestrations. - */ - version?: string; - - /** - * The default version to use when starting new orchestrations without an explicit version. - * This is used by the client when scheduling new orchestrations. - */ - defaultVersion?: string; - - /** - * The strategy for matching orchestration versions. - * @default VersionMatchStrategy.None - */ - matchStrategy?: VersionMatchStrategy; - - /** - * The strategy for handling version mismatches. - * @default VersionFailureStrategy.Reject - */ - failureStrategy?: VersionFailureStrategy; -} +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +/** + * Strategy for matching orchestration and activity versions when processing work items. + */ +export enum VersionMatchStrategy { + /** + * No version filtering. Local registration lookup still uses the requested version. + */ + None = 0, + + /** + * Only process work items that match the worker's version using version comparison. + */ + Strict = 1, + + /** + * Process work items with the current version or older versions. + * Uses semantic versioning comparison. + */ + CurrentOrOlder = 2, +} + +/** + * Strategy for handling version mismatches when processing work items. + */ +export enum VersionFailureStrategy { + /** + * Reject the work item and let it be picked up by another worker. + * The orchestration will be retried by a compatible worker. + */ + Reject = 0, + + /** + * Fail the orchestration with a version mismatch error. + * This will mark the orchestration as failed. + */ + Fail = 1, +} + +/** + * Options for version-based worker acceptance and default child orchestration versions. + */ +export interface VersioningOptions { + /** + * The version of the worker. This is used for version matching when processing orchestrations. + */ + version?: string; + + /** + * The default version for sub-orchestrations scheduled by this worker without an explicit + * version. An explicit empty string selects the unversioned child. This does not change + * the parent's version or activity versions. Top-level starts use the client's defaultVersion. + */ + defaultVersion?: string; + + /** + * The strategy for matching orchestration versions. + * @default VersionMatchStrategy.None + */ + matchStrategy?: VersionMatchStrategy; + + /** + * The strategy for handling version mismatches. + * @default VersionFailureStrategy.Reject + */ + failureStrategy?: VersionFailureStrategy; +} diff --git a/packages/durabletask-js/src/worker/work-item-filters.ts b/packages/durabletask-js/src/worker/work-item-filters.ts index f379bb4f..1ab4efe7 100644 --- a/packages/durabletask-js/src/worker/work-item-filters.ts +++ b/packages/durabletask-js/src/worker/work-item-filters.ts @@ -60,20 +60,27 @@ export function generateWorkItemFiltersFromRegistry( registry: Registry, versioning?: VersioningOptions, ): WorkItemFilters { - const versions: string[] = []; - if (versioning?.matchStrategy === VersionMatchStrategy.Strict && versioning.version) { - versions.push(versioning.version); + function taskFilters(registrations: { name: string; version: string }[]): OrchestrationWorkItemFilter[] { + const byName = new Map(); + for (const { name, version } of registrations) { + const versions = byName.get(name) ?? []; + versions.push(version); + byName.set(name, versions); + } + return [...byName].map(([name, versions]) => ({ + name, + versions: + versioning?.matchStrategy === VersionMatchStrategy.Strict + ? [versioning.version ?? ""] + : versions.length === 1 && versions[0] === "" + ? [] + : versions.sort(), + })); } return { - orchestrations: registry.getOrchestratorNames().map((name) => ({ - name, - versions: [...versions], - })), - activities: registry.getActivityNames().map((name) => ({ - name, - versions: [...versions], - })), + orchestrations: taskFilters(registry.getOrchestratorRegistrations()), + activities: taskFilters(registry.getActivityRegistrations()), entities: registry.getEntityNames().map((name) => ({ name, })), 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 0da19524..d5f720e8 100644 --- a/packages/durabletask-js/test/orchestration-id-reuse-policy.spec.ts +++ b/packages/durabletask-js/test/orchestration-id-reuse-policy.spec.ts @@ -307,7 +307,6 @@ describe("TestOrchestrationClient orchestration ID reuse policy", () => { 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, { diff --git a/packages/durabletask-js/test/versioned-dispatch-grpc.spec.ts b/packages/durabletask-js/test/versioned-dispatch-grpc.spec.ts new file mode 100644 index 00000000..333746be --- /dev/null +++ b/packages/durabletask-js/test/versioned-dispatch-grpc.spec.ts @@ -0,0 +1,239 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import * as grpc from "@grpc/grpc-js"; +import { Empty } from "google-protobuf/google/protobuf/empty_pb"; +import { NoOpLogger, TaskHubGrpcWorker, VersionMatchStrategy, VersionFailureStrategy, VersioningOptions } from "../src"; +import * as pb from "../src/proto/orchestrator_service_pb"; +import * as stubs from "../src/proto/orchestrator_service_grpc_pb"; +import * as ph from "../src/utils/pb-helper.util"; +import { withTimeout } from "../src/utils/backoff.util"; + +function deferred() { + let resolve!: (value: T) => void; + const promise = new Promise((complete) => (resolve = complete)); + return { promise, resolve }; +} + +type Response = pb.OrchestratorResponse | pb.ActivityResponse | "abandoned"; + +describe("Version dispatch over local gRPC", () => { + let server: grpc.Server; + let worker: TaskHubGrpcWorker; + const responses = new Map>>(); + let nextToken = 0; + + beforeEach(() => { + server = new grpc.Server(); + responses.clear(); + }); + + afterEach(async () => { + try { + if (worker?.["_isRunning"]) await worker.stop(); + } finally { + server.forceShutdown(); + } + }); + + async function start(register: (worker: TaskHubGrpcWorker) => void, versioning?: VersioningOptions) { + const connected = deferred>(); + const service = { + hello: (_call, callback) => callback(null, new Empty()), + getWorkItems: (call) => { + call.once("cancelled", () => call.end()); + call.write(new pb.WorkItem().setHealthping(new pb.HealthPing())); + connected.resolve(call); + }, + completeOrchestratorTask: (call, callback) => { + responses.get(call.request.getCompletiontoken())!.resolve(call.request); + callback(null, new pb.CompleteTaskResponse()); + }, + completeActivityTask: (call, callback) => { + responses.get(call.request.getCompletiontoken())!.resolve(call.request); + callback(null, new pb.CompleteTaskResponse()); + }, + abandonTaskOrchestratorWorkItem: (call, callback) => { + responses.get(call.request.getCompletiontoken())!.resolve("abandoned"); + callback(null, new pb.AbandonOrchestrationTaskResponse()); + }, + abandonTaskActivityWorkItem: (call, callback) => { + responses.get(call.request.getCompletiontoken())!.resolve("abandoned"); + callback(null, new pb.AbandonActivityTaskResponse()); + }, + } satisfies Partial; + server.addService(stubs.TaskHubSidecarServiceService, service); + const port = await new Promise((resolve, reject) => { + server.bindAsync("127.0.0.1:0", grpc.ServerCredentials.createInsecure(), (error, boundPort) => + error ? reject(error) : resolve(boundPort), + ); + }); + worker = new TaskHubGrpcWorker({ + hostAddress: `127.0.0.1:${port}`, + logger: new NoOpLogger(), + shutdownTimeoutMs: 500, + versioning, + workItemFilters: "auto", + }); + register(worker); + await worker.start(); + return withTimeout(connected.promise, 5000); + } + + async function send( + stream: grpc.ServerWritableStream, + kind: "orchestrator" | "activity", + name: string, + version?: string, + ): Promise { + const token = `token-${++nextToken}`; + const response = deferred(); + responses.set(token, response); + const item = new pb.WorkItem().setCompletiontoken(token); + if (kind === "activity") { + item.setActivityrequest( + new pb.ActivityRequest() + .setName(name) + .setTaskid(1) + .setVersion(version === undefined ? undefined : ph.getStringValue(version)) + .setOrchestrationinstance(new pb.OrchestrationInstance().setInstanceid("instance")), + ); + } else { + item.setOrchestratorrequest( + new pb.OrchestratorRequest() + .setInstanceid("instance") + .setNeweventsList([ + ph.newOrchestratorStartedEvent(new Date("2026-01-01T00:00:00Z")), + ph.newExecutionStartedEvent(name, "instance", undefined, undefined, undefined, version), + ]), + ); + } + stream.write(item); + return withTimeout(response.promise, 5000); + } + + function completion(response: Response) { + if (response === "abandoned") throw new Error("Unexpected abandon"); + if (response instanceof pb.ActivityResponse) { + return { result: response.getResult()?.getValue(), failure: response.getFailuredetails() }; + } + const action = response.getActionsList()[0].getCompleteorchestration(); + return { result: action?.getResult()?.getValue(), failure: action?.getFailuredetails() }; + } + + it("dispatches same-name versions, legacy fallback, and unknown versions without cross-version execution", async () => { + const calls: string[] = []; + const stream = await start((worker) => { + for (const version of ["", "v1", "v2"]) { + worker.addNamedOrchestrator( + "Work", + () => { + calls.push(`orch-${version}`); + return version; + }, + version, + ); + worker.addNamedActivity( + "Work", + () => { + calls.push(`act-${version}`); + return version; + }, + version, + ); + } + worker.addNamedOrchestrator("Legacy", () => "legacy"); + worker.addNamedActivity("Legacy", () => "legacy"); + }); + const filters = stream.request.getWorkitemfilters()!; + expect(filters.getOrchestrationsList().map((f) => [f.getName(), f.getVersionsList()])).toEqual([ + ["Work", ["", "v1", "v2"]], + ["Legacy", []], + ]); + expect(filters.getActivitiesList().map((f) => [f.getName(), f.getVersionsList()])).toEqual([ + ["Work", ["", "v1", "v2"]], + ["Legacy", []], + ]); + for (const kind of ["orchestrator", "activity"] as const) { + for (const version of ["", "v1", "v2"]) { + expect(completion(await send(stream, kind, "Work", version)).result).toBe(JSON.stringify(version)); + } + expect(completion(await send(stream, kind, "Legacy", "unknown")).result).toBe('"legacy"'); + const failure = completion(await send(stream, kind, "Work", "unknown")).failure; + expect(failure?.getErrortype()).toBe( + kind === "activity" ? "ActivityNotRegisteredError" : "OrchestratorNotRegisteredError", + ); + expect(failure?.getErrormessage()).toContain("unknown"); + expect(failure?.getIsnonretriable()).toBe(true); + } + expect(calls).toEqual(["orch-", "orch-v1", "orch-v2", "act-", "act-v1", "act-v2"]); + expect(() => worker.addNamedOrchestrator("Late", () => 1, "v1")).toThrow(/running/); + expect(() => worker.addNamedActivity("Late", () => 1, "v1")).toThrow(/running/); + }); + + it.each([VersionFailureStrategy.Fail, VersionFailureStrategy.Reject])( + "enforces worker acceptance before either exact or fallback dispatch with failure strategy %s", + async (failureStrategy) => { + const wrong = jest.fn(() => "wrong"); + const stream = await start( + (worker) => { + worker.addNamedOrchestrator("Exact", wrong, "v2"); + worker.addNamedActivity("Exact", wrong, "v2"); + worker.addNamedOrchestrator("Legacy", wrong); + worker.addNamedActivity("Legacy", wrong); + }, + { version: "v1", matchStrategy: VersionMatchStrategy.Strict, failureStrategy }, + ); + for (const kind of ["orchestrator", "activity"] as const) { + for (const name of ["Exact", "Legacy"]) { + const response = await send(stream, kind, name, "v2"); + if (failureStrategy === VersionFailureStrategy.Reject) expect(response).toBe("abandoned"); + else { + const failure = completion(response).failure; + expect(failure?.getErrortype()).toBe("VersionMismatch"); + expect(failure?.getIsnonretriable()).toBe(true); + } + } + } + expect(wrong).not.toHaveBeenCalled(); + }, + ); + + it("treats Strict with an omitted worker version as unversioned, not as no filtering", async () => { + const stream = await start( + (worker) => { + worker.addNamedOrchestrator("Work", () => "legacy"); + worker.addNamedActivity("Work", () => "legacy"); + }, + { matchStrategy: VersionMatchStrategy.Strict, failureStrategy: VersionFailureStrategy.Fail }, + ); + expect(stream.request.getWorkitemfilters()?.getActivitiesList()[0].getVersionsList()).toEqual([""]); + for (const kind of ["orchestrator", "activity"] as const) { + expect(completion(await send(stream, kind, "Work")).result).toBe('"legacy"'); + expect(completion(await send(stream, kind, "Work", "v1")).failure?.getErrortype()).toBe("VersionMismatch"); + } + }); + + it("keeps CurrentOrOlder acceptance independent of local implementation identity", async () => { + const stream = await start( + (worker) => { + for (const version of ["1", "2"]) { + worker.addNamedOrchestrator("Work", () => version, version); + worker.addNamedActivity("Work", () => version, version); + } + }, + { + version: "2", + matchStrategy: VersionMatchStrategy.CurrentOrOlder, + failureStrategy: VersionFailureStrategy.Fail, + }, + ); + for (const kind of ["orchestrator", "activity"] as const) { + expect(completion(await send(stream, kind, "Work", "1")).result).toBe('"1"'); + expect(completion(await send(stream, kind, "Work", "1.0")).failure?.getErrortype()).toBe( + kind === "activity" ? "ActivityNotRegisteredError" : "OrchestratorNotRegisteredError", + ); + expect(completion(await send(stream, kind, "Work", "3")).failure?.getErrortype()).toBe("VersionMismatch"); + } + }); +}); diff --git a/packages/durabletask-js/test/versioned-dispatch.spec.ts b/packages/durabletask-js/test/versioned-dispatch.spec.ts new file mode 100644 index 00000000..e3086523 --- /dev/null +++ b/packages/durabletask-js/test/versioned-dispatch.spec.ts @@ -0,0 +1,298 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { NoOpLogger, Task, TaskHubGrpcWorker, VersionMatchStrategy, RetryPolicy, RetryContext } from "../src"; +import { Registry } from "../src/worker/registry"; +import { ActivityExecutor } from "../src/worker/activity-executor"; +import { generateWorkItemFiltersFromRegistry } from "../src/worker/work-item-filters"; +import * as pb from "../src/proto/orchestrator_service_pb"; +import * as ph from "../src/utils/pb-helper.util"; + +describe.each(["orchestrator", "activity"] as const)("%s version registrations", (kind) => { + function setup() { + const registry = new Registry(); + const add = + kind === "orchestrator" ? registry.addNamedOrchestrator.bind(registry) : registry.addNamedActivity.bind(registry); + const get = kind === "orchestrator" ? registry.getOrchestrator.bind(registry) : registry.getActivity.bind(registry); + return { registry, add, get }; + } + + it("dispatches exact versions without changing case-sensitive task names", () => { + const { add, get } = setup(); + const v1 = () => 1; + const v2 = () => 2; + add("Work", v1, "V1"); + add("Work", v2, "v2"); + expect(get("Work", "v1")).toBe(v1); + expect(get("Work", "V2")).toBe(v2); + expect(get("work", "v1")).toBeUndefined(); + expect(get("Work")).toBeUndefined(); + expect(get("Work", "v3")).toBeUndefined(); + }); + + it("uses an unversioned fallback only while the name has no versioned registrations", () => { + const { add, get } = setup(); + const legacy = () => 0; + add("Work", legacy); + expect(get("Work", "v1")).toBe(legacy); + add("Work", () => 1, "v1"); + expect(get("Work", "v2")).toBeUndefined(); + expect(get("Work", "")).toBe(legacy); + expect(get("Work")).toBe(legacy); + }); + + it("rejects duplicate version identities and whitespace-only registration versions", () => { + const { add } = setup(); + add("Work", () => 1, "V1"); + expect(() => add("Work", () => 2, "v1")).toThrow(/already exists/); + add("Work", () => 0); + expect(() => add("Work", () => 0, "")).toThrow(/already exists/); + expect(() => add("Other", () => 0, " \t")).toThrow(/version/i); + }); + + it("does not equate numeric components or trim nonempty version strings", () => { + const { add, get } = setup(); + const one = () => 1; + const oneDotZero = () => 2; + const padded = () => 3; + add("Work", one, "1"); + add("Work", oneDotZero, "1.0"); + add("Work", padded, " 1 "); + expect(get("Work", "1")).toBe(one); + expect(get("Work", "1.0")).toBe(oneDotZero); + expect(get("Work", " 1 ")).toBe(padded); + }); + + it("normalizes null versions from JavaScript callers to unversioned", () => { + const { add, get } = setup(); + const legacy = () => 0; + Reflect.apply(add, undefined, ["Work", legacy, null]); + add("Work", () => 1, "v1"); + expect(get("Work")).toBe(legacy); + expect(Reflect.apply(get, undefined, ["Work", null])).toBe(legacy); + expect(() => add("Work", legacy, "")).toThrow(/already exists/); + }); + + it("enumerates every registration but deduplicates names and auto filters", () => { + const { registry, add } = setup(); + const fn = () => 1; + add("Work", fn, "v2"); + add("Work", fn, "v1"); + add("Work", fn); + add("Legacy", fn); + const registrations = + kind === "orchestrator" ? registry.getOrchestratorRegistrations() : registry.getActivityRegistrations(); + expect(registrations).toEqual([ + { name: "Work", version: "v2", fn }, + { name: "Work", version: "v1", fn }, + { name: "Work", version: "", fn }, + { name: "Legacy", version: "", fn }, + ]); + const names = kind === "orchestrator" ? registry.getOrchestratorNames() : registry.getActivityNames(); + expect(names).toEqual(["Work", "Legacy"]); + const key = kind === "orchestrator" ? "orchestrations" : "activities"; + expect(generateWorkItemFiltersFromRegistry(registry)[key]).toEqual([ + { name: "Work", versions: ["", "v1", "v2"] }, + { name: "Legacy", versions: [] }, + ]); + for (const version of ["v1", undefined]) { + expect( + generateWorkItemFiltersFromRegistry(registry, { + matchStrategy: VersionMatchStrategy.Strict, + version, + })[key], + ).toEqual([ + { name: "Work", versions: [version ?? ""] }, + { name: "Legacy", versions: [version ?? ""] }, + ]); + } + }); +}); + +describe("Versioned execution", () => { + const start = new Date("2026-01-01T00:00:00Z"); + + function request(name: string, version?: string) { + return new pb.OrchestratorRequest() + .setInstanceid("instance") + .setNeweventsList([ + ph.newOrchestratorStartedEvent(start), + ph.newExecutionStartedEvent(name, "instance", undefined, undefined, "execution", version), + ]); + } + + async function run(worker: TaskHubGrpcWorker, req: pb.OrchestratorRequest) { + return pb.OrchestratorResponse.deserializeBinary( + await worker.processOrchestratorRequest(req.serializeBinary()), + ).getActionsList(); + } + + it("selects the history version on initial execution and replay, not the worker default", async () => { + const worker = new TaskHubGrpcWorker({ + logger: new NoOpLogger(), + versioning: { version: "unrelated", defaultVersion: "v2" }, + }); + for (const version of ["v1", "v2"]) { + worker.addNamedOrchestrator( + "Flow", + async function* (ctx): AsyncGenerator, unknown[], unknown> { + const value = yield ctx.callActivity("Work"); + return [version, ctx.version, value]; + }, + version, + ); + } + for (const version of ["v1", "v2"]) { + const req = request("Flow", version); + const actions = await run(worker, req); + expect(actions).toHaveLength(1); + expect(actions[0].getScheduletask()?.getVersion()?.getValue()).toBe(version); + req + .setPasteventsList([...req.getNeweventsList(), ph.newTaskScheduledEvent(1, "Work")]) + .setNeweventsList([ph.newTaskCompletedEvent(1, '"result"')]); + const replay = await run(worker, req); + expect(replay[0].getCompleteorchestration()?.getResult()?.getValue()).toBe( + JSON.stringify([version, version, "result"]), + ); + } + }); + + it.each([undefined, "", "missing"])( + "fails a missing version %s instead of invoking another implementation", + async (version) => { + const worker = new TaskHubGrpcWorker({ logger: new NoOpLogger() }); + const wrong = jest.fn(() => "wrong"); + worker.addNamedOrchestrator("Flow", wrong, "v1"); + const actions = await run(worker, request("Flow", version)); + const failure = actions[0].getCompleteorchestration(); + expect(failure?.getOrchestrationstatus()).toBe(pb.OrchestrationStatus.ORCHESTRATION_STATUS_FAILED); + expect(failure?.getFailuredetails()?.getErrortype()).toBe("OrchestratorNotRegisteredError"); + expect(failure?.getFailuredetails()?.getErrormessage()).toContain(version ?? ""); + expect(wrong).not.toHaveBeenCalled(); + }, + ); + + it("normalizes whitespace-only wire versions to unversioned dispatch", async () => { + const worker = new TaskHubGrpcWorker({ logger: new NoOpLogger() }); + worker.addNamedOrchestrator("Flow", () => "unversioned"); + worker.addNamedOrchestrator("Flow", () => "v1", "v1"); + const actions = await run(worker, request("Flow", " \t")); + expect(actions[0].getCompleteorchestration()?.getResult()?.getValue()).toBe('"unversioned"'); + }); + + it.each([ + [undefined, undefined, ""], + ["child-default", undefined, "child-default"], + ["child-default", "override", "override"], + ["child-default", "", ""], + ["", undefined, ""], + ])("schedules children with default=%s explicit=%s as %s", async (defaultVersion, version, expected) => { + const worker = new TaskHubGrpcWorker({ logger: new NoOpLogger(), versioning: { defaultVersion } }); + worker.addNamedOrchestrator( + "Flow", + async function* (ctx) { + yield ctx.callSubOrchestrator("Child", undefined, { version }); + }, + "parent", + ); + const actions = await run(worker, request("Flow", "parent")); + expect(actions[0].getCreatesuborchestration()?.getVersion()?.getValue() ?? "").toBe(expected); + }); + + it.each([undefined, "override", ""])("schedules activities with explicit version %s", async (version) => { + const worker = new TaskHubGrpcWorker({ + logger: new NoOpLogger(), + versioning: { defaultVersion: "not-the-activity-default" }, + }); + worker.addNamedOrchestrator( + "Flow", + async function* (ctx) { + yield ctx.callActivity("Work", undefined, { version }); + }, + "parent", + ); + const actions = await run(worker, request("Flow", "parent")); + expect(actions[0].getScheduletask()?.getVersion()?.getValue() ?? "").toBe(version ?? "parent"); + }); + + it("does not apply the child default to continue-as-new", async () => { + const worker = new TaskHubGrpcWorker({ logger: new NoOpLogger(), versioning: { defaultVersion: "v2" } }); + worker.addNamedOrchestrator("Flow", (ctx) => ctx.continueAsNew(1, false), "v1"); + const actions = await run(worker, request("Flow", "v1")); + expect(actions[0].getCompleteorchestration()?.hasNewversion()).toBe(false); + }); + + it.each(["policy", "handler"] as const)( + "retains default, explicit and empty child versions through %s retry replay", + async (kind) => { + for (const version of [undefined, "override", ""]) { + const worker = new TaskHubGrpcWorker({ + logger: new NoOpLogger(), + versioning: { defaultVersion: "child-default" }, + }); + const retry = + kind === "policy" + ? new RetryPolicy({ maxNumberOfAttempts: 2, firstRetryIntervalInMilliseconds: 1 }) + : (ctx: RetryContext) => ctx.lastAttemptNumber < 2; + worker.addNamedOrchestrator( + "Flow", + async function* (ctx) { + yield ctx.callSubOrchestrator("Child", undefined, { version, retry }); + return "done"; + }, + "parent", + ); + const req = request("Flow", "parent"); + const initial = (await run(worker, req))[0]; + const childId = initial.getCreatesuborchestration()!.getInstanceid(); + req + .setPasteventsList([ + ...req.getNeweventsList(), + ph.newSubOrchestrationCreatedEvent(initial.getId(), "Child", childId), + ]) + .setNeweventsList([ph.newSubOrchestrationFailedEvent(initial.getId(), new Error("retry"))]); + let actions = await run(worker, req); + if (kind === "policy") { + const timer = actions[0]; + const fireAt = timer.getCreatetimer()!.getFireat()!.toDate(); + req + .setPasteventsList([ + ...req.getPasteventsList(), + ...req.getNeweventsList(), + ph.newTimerCreatedEvent(timer.getId(), fireAt), + ]) + .setNeweventsList([ph.newOrchestratorStartedEvent(fireAt), ph.newTimerFiredEvent(timer.getId(), fireAt)]); + actions = await run(worker, req); + } + expect(actions).toHaveLength(1); + const retried = actions[0]; + expect(retried.getCreatesuborchestration()?.getVersion()?.getValue() ?? "").toBe(version ?? "child-default"); + expect(retried.getCreatesuborchestration()?.getInstanceid()).toBe(childId); + req + .setPasteventsList([ + ...req.getPasteventsList(), + ...req.getNeweventsList(), + ph.newSubOrchestrationCreatedEvent(retried.getId(), "Child", childId), + ]) + .setNeweventsList([ph.newSubOrchestrationCompletedEvent(retried.getId(), '"child"')]); + const completed = await run(worker, req); + expect(completed[0].getCompleteorchestration()?.getResult()?.getValue()).toBe('"done"'); + } + }, + ); + + it("dispatches activities by version and reports missing versions with the standard error", async () => { + const registry = new Registry(); + registry.addNamedActivity("Work", () => "unversioned"); + registry.addNamedActivity("Work", () => "v1", "v1"); + registry.addNamedActivity("Work", () => "v2", "v2"); + const executor = new ActivityExecutor(registry, new NoOpLogger()); + for (const version of ["", "v1", "v2"]) { + await expect(executor.execute("instance", "Work", 1, undefined, version)).resolves.toBe( + JSON.stringify(version || "unversioned"), + ); + } + await expect(executor.execute("instance", "Work", 1, undefined, "missing")).rejects.toThrow(/Work.*missing/); + await expect(executor.execute("instance", "Work", 1, undefined, " \t")).resolves.toBe('"unversioned"'); + }); +}); diff --git a/packages/durabletask-js/test/versioned-in-memory.spec.ts b/packages/durabletask-js/test/versioned-in-memory.spec.ts new file mode 100644 index 00000000..3b3cdb60 --- /dev/null +++ b/packages/durabletask-js/test/versioned-in-memory.spec.ts @@ -0,0 +1,160 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { + InMemoryOrchestrationBackend, + TestOrchestrationClient, + TestOrchestrationWorker, + OrchestrationStatus, + OrchestrationContext, + RetryPolicy, + RetryContext, + Task, +} from "../src"; + +describe("Versioned in-memory workflows", () => { + let backend: InMemoryOrchestrationBackend; + let client: TestOrchestrationClient; + let worker: TestOrchestrationWorker; + + beforeEach(() => { + backend = new InMemoryOrchestrationBackend(); + client = new TestOrchestrationClient(backend); + worker = new TestOrchestrationWorker(backend, { versioning: { defaultVersion: "v2" } }); + }); + + afterEach(async () => { + await worker.stop(); + backend.reset(); + }); + + it("preserves inferred function names across versioned test-worker registrations", async () => { + for (const version of ["v1", "v2"]) { + async function* Flow(ctx: OrchestrationContext): AsyncGenerator, unknown, unknown> { + return yield ctx.callActivity("Work"); + } + function Work() { + return version; + } + expect(worker.addOrchestrator(Flow, version)).toBe("Flow"); + expect(worker.addActivity(Work, version)).toBe("Work"); + } + await worker.start(); + for (const version of ["v1", "v2"]) { + const id = await client.scheduleNewOrchestration("Flow", undefined, { version }); + const state = await client.waitForOrchestrationCompletion(id, true, 5); + expect(state?.runtimeStatus).toBe(OrchestrationStatus.COMPLETED); + expect(state?.serializedOutput).toBe(JSON.stringify(version)); + } + }); + + it("runs same-name orchestrations and activities side by side through replay", async () => { + for (const version of ["v1", "v2"]) { + worker.addNamedOrchestrator( + "Flow", + async function* (ctx): AsyncGenerator, unknown[], unknown> { + const activity = yield ctx.callActivity("Work"); + const child = yield ctx.callSubOrchestrator("Child"); + const explicit = yield ctx.callSubOrchestrator("Child", undefined, { version: "v1" }); + const unversioned = yield ctx.callSubOrchestrator("Child", undefined, { version: "" }); + return [ctx.version, activity, child, explicit, unversioned]; + }, + version, + ); + worker.addNamedOrchestrator("Child", (ctx) => [version, ctx.version], version); + worker.addNamedActivity("Work", () => version, version); + } + worker.addNamedOrchestrator("Child", (ctx) => ["legacy", ctx.version]); + await worker.start(); + for (const version of ["v1", "v2"]) { + const id = await client.scheduleNewOrchestration("Flow", undefined, { version }); + const state = await client.waitForOrchestrationCompletion(id, true, 5); + expect(state?.runtimeStatus).toBe(OrchestrationStatus.COMPLETED); + expect(state?.serializedOutput).toBe( + JSON.stringify([version, version, ["v2", "v2"], ["v1", "v1"], ["legacy", ""]]), + ); + const history = backend.getInstance(id)!.history; + expect( + history + .find((event) => event.hasTaskscheduled()) + ?.getTaskscheduled() + ?.getVersion() + ?.getValue(), + ).toBe(version); + expect( + history + .filter((event) => event.hasSuborchestrationinstancecreated()) + .map((event) => event.getSuborchestrationinstancecreated()?.getVersion()?.getValue() ?? ""), + ).toEqual(["v2", "v1", ""]); + } + }); + + it.each(["policy", "handler"] as const)("retains activity versions through %s retries", async (kind) => { + let attempts = 0; + const retry = + kind === "policy" + ? new RetryPolicy({ maxNumberOfAttempts: 2, firstRetryIntervalInMilliseconds: 1 }) + : (ctx: RetryContext) => ctx.lastAttemptNumber < 2; + worker.addNamedOrchestrator("Flow", async function* (ctx): AsyncGenerator, unknown[], unknown> { + const activity = yield ctx.callActivity("Work", undefined, { version: "v1", retry }); + return [activity]; + }); + worker.addNamedActivity( + "Work", + () => { + if (++attempts === 1) throw new Error("retry activity"); + return "activity-v1"; + }, + "v1", + ); + worker.addNamedActivity( + "Work", + () => { + throw new Error("wrong activity version"); + }, + "v2", + ); + await worker.start(); + const id = await client.scheduleNewOrchestration("Flow"); + const state = await client.waitForOrchestrationCompletion(id, true, 5); + expect(state?.runtimeStatus).toBe(OrchestrationStatus.COMPLETED); + expect(state?.serializedOutput).toBe('["activity-v1"]'); + expect(attempts).toBe(2); + }); + + it("does not dispatch an unknown version to the unversioned implementation after opting into versions", async () => { + const wrong = jest.fn(() => "wrong"); + worker.addNamedOrchestrator("Flow", wrong); + worker.addNamedOrchestrator("Flow", wrong, "v1"); + await worker.start(); + const id = await client.scheduleNewOrchestration("Flow", undefined, { version: "v3" }); + const state = await client.waitForOrchestrationCompletion(id, true, 5); + expect(state?.runtimeStatus).toBe(OrchestrationStatus.FAILED); + expect(wrong).not.toHaveBeenCalled(); + }); + + it("reports missing activity versions as non-retriable failures", async () => { + const wrong = jest.fn(() => "wrong"); + worker.addNamedOrchestrator("Flow", async function* (ctx) { + yield ctx.callActivity("Work", undefined, { + version: "missing", + retry: new RetryPolicy({ maxNumberOfAttempts: 2, firstRetryIntervalInMilliseconds: 1 }), + }); + }); + worker.addNamedActivity("Work", wrong, "v1"); + await worker.start(); + const id = await client.scheduleNewOrchestration("Flow"); + const state = await client.waitForOrchestrationCompletion(id, true, 5); + expect(state?.runtimeStatus).toBe(OrchestrationStatus.FAILED); + expect(wrong).not.toHaveBeenCalled(); + const history = backend.getInstance(id)!.history; + expect(history.filter((event) => event.hasTaskscheduled())).toHaveLength(1); + expect( + history + .find((event) => event.hasTaskfailed()) + ?.getTaskfailed() + ?.getFailuredetails() + ?.getIsnonretriable(), + ).toBe(true); + }); +}); diff --git a/packages/durabletask-js/test/work-item-filters.spec.ts b/packages/durabletask-js/test/work-item-filters.spec.ts index 49b72839..004f5c3a 100644 --- a/packages/durabletask-js/test/work-item-filters.spec.ts +++ b/packages/durabletask-js/test/work-item-filters.spec.ts @@ -402,7 +402,7 @@ describe("WorkItemFilters", () => { expect(filters.activities![0].versions).toEqual([]); }); - it("should not include version when Strict but version is not set", () => { + it("should filter for unversioned work when Strict but version is not set", () => { // Arrange const registry = new Registry(); registry.addOrchestrator(myOrchestrator); @@ -413,7 +413,7 @@ describe("WorkItemFilters", () => { }); // Assert - expect(filters.orchestrations![0].versions).toEqual([]); + expect(filters.orchestrations![0].versions).toEqual([""]); }); it("should not include version when no versioning options provided", () => { From 27c167bcd1c764162eeaf9c0fefaea27290ec965 Mon Sep 17 00:00:00 2001 From: wangbill Date: Tue, 22 Sep 2026 14:34:25 -0700 Subject: [PATCH 2/2] fix: preserve empty task versions on the wire Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: e0af01a5-0dfa-4e71-a660-c4186e65d7e0 --- CHANGELOG.md | 2 ++ README.md | 2 ++ .../src/utils/pb-helper.util.ts | 8 ++----- .../worker/runtime-orchestration-context.ts | 2 +- .../test/pb-helper-versioning.spec.ts | 22 ++++++++++++++++++- .../test/versioned-dispatch.spec.ts | 3 +++ 6 files changed, 31 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 80a550bc..fa7c3a65 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,6 +25,8 @@ ### Fixes +- Preserve explicit empty activity and child versions on the protobuf wire, including the + unversioned child default, so concrete empty-version work-item filters can match them. - Reject uninitialized `whenAll` results after a canceled child's completion callback throws, rather than allowing a caught cancellation followed by `yield` to report success. - Align worker response cancellation with .NET: `stop()` cancels initial sends as well diff --git a/README.md b/README.md index e26572c1..c09de53c 100644 --- a/README.md +++ b/README.md @@ -311,6 +311,8 @@ different registrations. Omitted, `undefined`, `null` (JavaScript), and `""` ver unversioned registration. Whitespace-only registration versions are rejected; other strings are not trimmed. Duplicate name/version pairs throw. **Task names remain case-sensitive in JavaScript**, unlike .NET; this preserves existing JavaScript identifiers and function-name behavior. +Backend constraints still apply: Azure DTS accepts numeric `Major[.Minor[.Patch]]` orchestration +versions, so use values such as `"1.0.0"` and `"2.0.0"` rather than `"v1"` and `"v2"` with that service. Dispatch uses the recorded `ExecutionStarted.version` during both initial execution and replay, and `ActivityRequest.version` for activities. Exact matches win. Following the diff --git a/packages/durabletask-js/src/utils/pb-helper.util.ts b/packages/durabletask-js/src/utils/pb-helper.util.ts index 8cb61000..a62fba5c 100644 --- a/packages/durabletask-js/src/utils/pb-helper.util.ts +++ b/packages/durabletask-js/src/utils/pb-helper.util.ts @@ -454,9 +454,7 @@ export function newScheduleTaskAction( scheduleTaskAction.setName(name); scheduleTaskAction.setInput(getStringValue(encodedInput)); populateTagsMap(scheduleTaskAction.getTagsMap(), tags); - if (version) { - scheduleTaskAction.setVersion(getStringValue(version)); - } + scheduleTaskAction.setVersion(getStringValueIfDefined(version)); const action = new pb.OrchestratorAction(); action.setId(id); @@ -484,9 +482,7 @@ export function newCreateSubOrchestrationAction( createSubOrchestrationAction.setInstanceid(instanceId || ""); createSubOrchestrationAction.setInput(getStringValue(encodedInput)); populateTagsMap(createSubOrchestrationAction.getTagsMap(), tags); - if (version) { - createSubOrchestrationAction.setVersion(getStringValue(version)); - } + createSubOrchestrationAction.setVersion(getStringValueIfDefined(version)); const action = new pb.OrchestratorAction(); action.setId(id); diff --git a/packages/durabletask-js/src/worker/runtime-orchestration-context.ts b/packages/durabletask-js/src/worker/runtime-orchestration-context.ts index 4c45e0cc..eddf048b 100644 --- a/packages/durabletask-js/src/worker/runtime-orchestration-context.ts +++ b/packages/durabletask-js/src/worker/runtime-orchestration-context.ts @@ -470,7 +470,7 @@ export class RuntimeOrchestrationContext extends OrchestrationContext { instanceId, encodedInput, options?.tags, - options?.version ?? this._defaultVersion, + options?.version ?? this._defaultVersion ?? "", ); this._pendingActions[action.getId()] = action; diff --git a/packages/durabletask-js/test/pb-helper-versioning.spec.ts b/packages/durabletask-js/test/pb-helper-versioning.spec.ts index 2fa544b0..0b9ada82 100644 --- a/packages/durabletask-js/test/pb-helper-versioning.spec.ts +++ b/packages/durabletask-js/test/pb-helper-versioning.spec.ts @@ -1,7 +1,27 @@ // Copyright (c) Microsoft Corporation. All rights reserved. // Licensed under the MIT License. -import { newVersionMismatchFailureDetails, newFailureDetails } from "../src/utils/pb-helper.util"; +import { + newVersionMismatchFailureDetails, + newFailureDetails, + newScheduleTaskAction, + newCreateSubOrchestrationAction, +} from "../src/utils/pb-helper.util"; +import { OrchestratorAction } from "../src/proto/orchestrator_service_pb"; + +describe.each(["activity", "child"] as const)("serialized %s version presence", (kind) => { + it.each([undefined, "", "1.0.0"])("preserves the distinction between omitted and explicit version %s", (version) => { + const action = + kind === "activity" + ? newScheduleTaskAction(1, "Work", undefined, undefined, version) + : newCreateSubOrchestrationAction(1, "Work", "child", undefined, undefined, version); + const decoded = OrchestratorAction.deserializeBinary(action.serializeBinary()); + const task = kind === "activity" ? decoded.getScheduletask()! : decoded.getCreatesuborchestration()!; + + expect(task.hasVersion()).toBe(version !== undefined); + expect(task.getVersion()?.getValue()).toBe(version); + }); +}); describe("pb-helper.util - Version Mismatch Failure Details", () => { describe("newVersionMismatchFailureDetails", () => { diff --git a/packages/durabletask-js/test/versioned-dispatch.spec.ts b/packages/durabletask-js/test/versioned-dispatch.spec.ts index e3086523..df4b7551 100644 --- a/packages/durabletask-js/test/versioned-dispatch.spec.ts +++ b/packages/durabletask-js/test/versioned-dispatch.spec.ts @@ -196,6 +196,7 @@ describe("Versioned execution", () => { "parent", ); const actions = await run(worker, request("Flow", "parent")); + expect(actions[0].getCreatesuborchestration()?.hasVersion()).toBe(true); expect(actions[0].getCreatesuborchestration()?.getVersion()?.getValue() ?? "").toBe(expected); }); @@ -212,6 +213,7 @@ describe("Versioned execution", () => { "parent", ); const actions = await run(worker, request("Flow", "parent")); + expect(actions[0].getScheduletask()?.hasVersion()).toBe(true); expect(actions[0].getScheduletask()?.getVersion()?.getValue() ?? "").toBe(version ?? "parent"); }); @@ -266,6 +268,7 @@ describe("Versioned execution", () => { } expect(actions).toHaveLength(1); const retried = actions[0]; + expect(retried.getCreatesuborchestration()?.hasVersion()).toBe(true); expect(retried.getCreatesuborchestration()?.getVersion()?.getValue() ?? "").toBe(version ?? "child-default"); expect(retried.getCreatesuborchestration()?.getInstanceid()).toBe(childId); req