diff --git a/.github/workflows/functions-e2e-tests.yaml b/.github/workflows/functions-e2e-tests.yaml index 7f62603c..2f23cc22 100644 --- a/.github/workflows/functions-e2e-tests.yaml +++ b/.github/workflows/functions-e2e-tests.yaml @@ -64,8 +64,14 @@ jobs: - name: "🏗️ Build in-repo durable-functions (+ core) for file: linking" run: npm run build -w durable-functions + # Core Tools is pinned to an exact version rather than the floating `@4` + # range: its postinstall downloads a platform zip from cdn.functions.azure.com + # and hard-fails (process.exit(1)) on any non-200, so a bad upstream publish + # breaks this job for every PR. `@4` floated onto 4.13.2, whose linux-x64 zip + # 404s on the CDN. Bump this pin deliberately once a newer version is + # confirmed to install. - name: 🔧 Install Azurite and Azure Functions Core Tools - run: npm install -g azurite azure-functions-core-tools@4 + run: npm install -g azurite azure-functions-core-tools@4.12.1 # --skipApiVersionCheck: the preview extension bundle's Azure Storage SDK # targets a newer REST API version than current Azurite accepts; without the diff --git a/CHANGELOG.md b/CHANGELOG.md index 46e44627..b4508367 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ### New +- Add an optional `newVersion` parameter to `OrchestrationContext.continueAsNew()` for version migrations. - Implement entity support in the in-memory testing backend ([#341](https://github.com/microsoft/durabletask-js/pull/341)) - Add the top-level `StartOrchestrationOptions.dedupeStatuses` option, `ValidDedupeStatuses`, and `OrchestrationAlreadyExistsError`, aligned with the .NET status-based duplicate rejection and diff --git a/README.md b/README.md index aeaf9350..9c5b9ef9 100644 --- a/README.md +++ b/README.md @@ -170,6 +170,25 @@ 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). +### Continue as new + +Long-running orchestrations can restart with fresh history and optionally move to a new +orchestration version: + +```typescript +const eternalOrchestrator: TOrchestrator = async function* ( + ctx: OrchestrationContext, + iteration: number, +): any { + yield ctx.callActivity(processIteration, iteration); + ctx.continueAsNew(iteration + 1, true, "2.0.0"); +}; +``` + +The second argument controls whether unprocessed external events carry over. The optional third +argument becomes the restarted orchestration's `ctx.version`; omit it to retain the existing +continue-as-new behavior. + ### Durable entities Durable entities provide a way to manage small pieces of state with a simple object-oriented programming model: diff --git a/packages/azure-functions-durable/CHANGELOG.md b/packages/azure-functions-durable/CHANGELOG.md index 880f1bd6..20f5d7f9 100644 --- a/packages/azure-functions-durable/CHANGELOG.md +++ b/packages/azure-functions-durable/CHANGELOG.md @@ -2,6 +2,7 @@ ### New +- Add optional orchestration version migration support to `context.df.continueAsNew()`. - Added a `durable-functions/testing` entry point with `runOrchestrator`, which runs an orchestrator to a terminal state against inline activity implementations on the in-memory backend and always releases its worker, and `createActivityContext` for invoking activity handlers directly. diff --git a/packages/azure-functions-durable/README.md b/packages/azure-functions-durable/README.md index d672a302..a8ec7ee9 100644 --- a/packages/azure-functions-durable/README.md +++ b/packages/azure-functions-durable/README.md @@ -37,6 +37,10 @@ changed: and throws when the instance is missing. `showInput` suppresses only the top-level input, `showHistory` populates `history`, and `showHistoryOutput` toggles the per-entry input/result payloads; `history` entries are core `HistoryEvent`s (v3 types `history` as `Array`). + **`client.startNew()` supports the `version` option.** +- **`context.df.continueAsNew(input, saveEvents, newVersion)` can migrate versions.** The optional + third argument assigns the restarted orchestration's version; existing one- and two-argument + calls keep their current behavior. **`client.startNew()` supports the `version` and top-level `dedupeStatuses` options.** Dedupe statuses select duplicate errors; other supported statuses are atomically replaced. The current shared protocol does not expose an atomic no-op/`IGNORE` action. diff --git a/packages/azure-functions-durable/src/orchestration-context.ts b/packages/azure-functions-durable/src/orchestration-context.ts index b4102e3b..4349d521 100644 --- a/packages/azure-functions-durable/src/orchestration-context.ts +++ b/packages/azure-functions-durable/src/orchestration-context.ts @@ -127,9 +127,13 @@ export class DurableOrchestrationContext { return this._ctx.waitForExternalEvent(name) as Task; } - /** Restarts the orchestration with a new input. */ - continueAsNew(input: unknown, saveEvents = true): void { - this._ctx.continueAsNew(input, saveEvents); + /** Restarts the orchestration with a new input and optionally a new version. */ + continueAsNew(input: unknown, saveEvents = true, newVersion?: string): void { + if (newVersion === undefined) { + this._ctx.continueAsNew(input, saveEvents); + } else { + this._ctx.continueAsNew(input, saveEvents, newVersion); + } } /** Sets the orchestration's custom status payload. */ diff --git a/packages/azure-functions-durable/test/unit/orchestration-context.spec.ts b/packages/azure-functions-durable/test/unit/orchestration-context.spec.ts index eb8f7e2d..2f9f7289 100644 --- a/packages/azure-functions-durable/test/unit/orchestration-context.spec.ts +++ b/packages/azure-functions-durable/test/unit/orchestration-context.spec.ts @@ -151,6 +151,15 @@ describe("DurableOrchestrationContext", () => { expect(entities.signalEntity).toHaveBeenCalledWith(entityId, "reset", undefined); }); + it("forwards a new version when continuing as new", () => { + const { ctx, raw } = createFakeCoreContext(); + const df = new DurableOrchestrationContext(ctx, undefined); + + df.continueAsNew("next", false, "2.0.0"); + + expect(raw.continueAsNew).toHaveBeenCalledWith("next", false, "2.0.0"); + }); + it("schedules callHttp as the built-in poll sub-orchestration with the built request payload", () => { const { ctx, raw } = createFakeCoreContext(); const df = new DurableOrchestrationContext(ctx, undefined); diff --git a/packages/durabletask-js/src/orchestration/exception/orchestration-already-exists-error.ts b/packages/durabletask-js/src/orchestration/exception/orchestration-already-exists-error.ts index c50f22fc..a0e2eda6 100644 --- a/packages/durabletask-js/src/orchestration/exception/orchestration-already-exists-error.ts +++ b/packages/durabletask-js/src/orchestration/exception/orchestration-already-exists-error.ts @@ -3,7 +3,11 @@ /** Thrown when an orchestration ID reuse request matches an existing dedupe status. */ export class OrchestrationAlreadyExistsError extends Error { - constructor(message: string, options?: ErrorOptions) { + // The options type is spelled out structurally instead of using the ambient + // `ErrorOptions`, which only exists in the ES2022 lib. Naming it here would leak + // into the emitted .d.ts and break consumers compiling against an older lib + // (e.g. the `func`-style apps that default to `target: es6` without skipLibCheck). + constructor(message: string, options?: { cause?: unknown }) { super(message, options); this.name = "OrchestrationAlreadyExistsError"; } diff --git a/packages/durabletask-js/src/task/context/orchestration-context.ts b/packages/durabletask-js/src/task/context/orchestration-context.ts index bf75699f..3847a0d7 100644 --- a/packages/durabletask-js/src/task/context/orchestration-context.ts +++ b/packages/durabletask-js/src/task/context/orchestration-context.ts @@ -163,8 +163,9 @@ export abstract class OrchestrationContext { * * @param newInput {any} The new input to use for the new orchestration instance. * @param saveEvents {boolean} A flag indicating whether to add any unprocessed external events in the new orchestration history. + * @param newVersion {string} The optional version to use for the new orchestration instance. */ - abstract continueAsNew(newInput: any, saveEvents: boolean): void; + abstract continueAsNew(newInput: any, saveEvents: boolean, newVersion?: string): void; /** * Sets a custom status value for the current orchestration instance. diff --git a/packages/durabletask-js/src/testing/in-memory-backend.ts b/packages/durabletask-js/src/testing/in-memory-backend.ts index a0d9a498..6937381a 100644 --- a/packages/durabletask-js/src/testing/in-memory-backend.ts +++ b/packages/durabletask-js/src/testing/in-memory-backend.ts @@ -1049,6 +1049,14 @@ export class InMemoryOrchestrationBackend { if (status === pb.OrchestrationStatus.ORCHESTRATION_STATUS_CONTINUED_AS_NEW) { // Handle continue-as-new const newInput = completeAction.getResult()?.getValue(); + const currentVersion = instance.history + .find((event) => event.hasExecutionstarted()) + ?.getExecutionstarted() + ?.getVersion() + ?.getValue(); + const newVersion = completeAction.hasNewversion() + ? completeAction.getNewversion()?.getValue() + : currentVersion; const carryoverEvents = completeAction.getCarryovereventsList(); // Cancel timers still pending from the previous iteration. Their timer IDs are @@ -1084,6 +1092,7 @@ export class InMemoryOrchestrationBackend { newInput, undefined, instance.executionId, + newVersion ); instance.pendingEvents = [orchestratorStarted, executionStarted, ...carryoverEvents]; diff --git a/packages/durabletask-js/src/utils/pb-helper.util.ts b/packages/durabletask-js/src/utils/pb-helper.util.ts index 9f353798..45e500cb 100644 --- a/packages/durabletask-js/src/utils/pb-helper.util.ts +++ b/packages/durabletask-js/src/utils/pb-helper.util.ts @@ -22,7 +22,14 @@ export function newOrchestratorStartedEvent(timestamp?: Date | null): pb.History return event; } -export function newExecutionStartedEvent(name: string, instanceId: string, encodedInput?: string, parentInstance?: { name: string; instanceId: string; taskScheduledId: number }, executionId?: string): pb.HistoryEvent { +export function newExecutionStartedEvent( + name: string, + instanceId: string, + encodedInput?: string, + parentInstance?: { name: string; instanceId: string; taskScheduledId: number }, + executionId?: string, + version?: string, +): pb.HistoryEvent { const ts = new Timestamp(); const orchestrationInstance = new pb.OrchestrationInstance(); @@ -39,6 +46,7 @@ export function newExecutionStartedEvent(name: string, instanceId: string, encod executionStartedEvent.setName(name); executionStartedEvent.setInput(getStringValue(encodedInput)); executionStartedEvent.setOrchestrationinstance(orchestrationInstance); + executionStartedEvent.setVersion(getStringValueIfDefined(version)); // Set parent instance info if provided (for sub-orchestrations) if (parentInstance) { @@ -360,6 +368,16 @@ export function getStringValue(val?: string): StringValue | undefined { return stringValue; } +function getStringValueIfDefined(val?: string): StringValue | undefined { + if (val === undefined) { + return; + } + + const stringValue = new StringValue(); + stringValue.setValue(val); + return stringValue; +} + /** * Populates a tag map with the provided tags. * @@ -390,12 +408,14 @@ export function newCompleteOrchestrationAction( result?: string, failureDetails?: pb.TaskFailureDetails, carryoverEvents?: pb.HistoryEvent[] | null, + newVersion?: string, ): pb.OrchestratorAction { const completeOrchestrationAction = new pb.CompleteOrchestrationAction(); completeOrchestrationAction.setOrchestrationstatus(status); completeOrchestrationAction.setResult(getStringValue(result)); completeOrchestrationAction.setFailuredetails(failureDetails); completeOrchestrationAction.setCarryovereventsList(carryoverEvents || []); + completeOrchestrationAction.setNewversion(getStringValueIfDefined(newVersion)); const action = new pb.OrchestratorAction(); action.setId(id); @@ -679,4 +699,3 @@ function wrapEntityMessageAction( action.setSendentitymessage(sendEntityMessage); return action; } - diff --git a/packages/durabletask-js/src/worker/runtime-orchestration-context.ts b/packages/durabletask-js/src/worker/runtime-orchestration-context.ts index 828deb4e..39caea41 100644 --- a/packages/durabletask-js/src/worker/runtime-orchestration-context.ts +++ b/packages/durabletask-js/src/worker/runtime-orchestration-context.ts @@ -49,6 +49,7 @@ export class RuntimeOrchestrationContext extends OrchestrationContext { _pendingEvents: Record[]>; _newInput?: any; _saveEvents: boolean; + _newVersion?: string; _customStatus?: string; _entityFeature: RuntimeOrchestrationEntityFeature; @@ -72,6 +73,7 @@ export class RuntimeOrchestrationContext extends OrchestrationContext { this._pendingEvents = {}; this._newInput = undefined; this._saveEvents = false; + this._newVersion = undefined; this._customStatus = undefined; this._entityFeature = new RuntimeOrchestrationEntityFeature(this); } @@ -258,7 +260,7 @@ export class RuntimeOrchestrationContext extends OrchestrationContext { this._pendingActions[action.getId()] = action; } - setContinuedAsNew(newInput: any, saveEvents: boolean) { + setContinuedAsNew(newInput: any, saveEvents: boolean, newVersion?: string) { if (this._isComplete) { return; } @@ -267,6 +269,7 @@ export class RuntimeOrchestrationContext extends OrchestrationContext { this._completionStatus = pb.OrchestrationStatus.ORCHESTRATION_STATUS_CONTINUED_AS_NEW; this._newInput = newInput; this._saveEvents = saveEvents; + this._newVersion = newVersion; } getActions(): pb.OrchestratorAction[] { @@ -292,6 +295,7 @@ export class RuntimeOrchestrationContext extends OrchestrationContext { this._newInput !== undefined ? JSON.stringify(this._newInput) : undefined, undefined, carryoverEvents, + this._newVersion, ); // Include fire-and-forget actions (sendEvent, signalEntity, etc.) that were @@ -459,14 +463,15 @@ export class RuntimeOrchestrationContext extends OrchestrationContext { } /** - * Orchestrations can be continued as new. This API allows an orchestration to restart itself from scratch, optionally with a new input. + * Restarts the orchestration with fresh history, optionally carrying over unprocessed events + * and assigning a new orchestration version. */ - continueAsNew(newInput: any, saveEvents: boolean = false) { + continueAsNew(newInput: any, saveEvents: boolean = false, newVersion?: string) { if (this._isComplete) { return; } - this.setContinuedAsNew(newInput, saveEvents); + this.setContinuedAsNew(newInput, saveEvents, newVersion); } /** diff --git a/packages/durabletask-js/test/in-memory-backend.spec.ts b/packages/durabletask-js/test/in-memory-backend.spec.ts index 647845ec..0c9f11ac 100644 --- a/packages/durabletask-js/test/in-memory-backend.spec.ts +++ b/packages/durabletask-js/test/in-memory-backend.spec.ts @@ -14,6 +14,7 @@ import { TOrchestrator, } from "../src"; import * as pb from "../src/proto/orchestrator_service_pb"; +import { StringValue } from "google-protobuf/google/protobuf/wrappers_pb"; describe("In-Memory Backend", () => { let backend: InMemoryOrchestrationBackend; @@ -331,6 +332,81 @@ describe("In-Memory Backend", () => { expect(state?.serializedOutput).toEqual(JSON.stringify(5)); }); + it("should retain the current version when continue-as-new omits a new version", async () => { + const observedVersions: string[] = []; + const orchestrator: TOrchestrator = async (ctx: OrchestrationContext, input: number) => { + if (!ctx.isReplaying) { + observedVersions.push(ctx.version); + } + + if (input === 0) { + ctx.continueAsNew(1, false, "2.0.0"); + return; + } + + if (input === 1) { + ctx.continueAsNew(2, false); + return; + } + + return ctx.version; + }; + + worker.addOrchestrator(orchestrator); + const id = "versioned-continue-as-new"; + backend.createInstance(id, getName(orchestrator), JSON.stringify(0)); + const initialExecutionStarted = backend + .getInstance(id) + ?.pendingEvents.find((event) => event.hasExecutionstarted()) + ?.getExecutionstarted(); + const initialVersion = new StringValue(); + initialVersion.setValue("1.0.0"); + initialExecutionStarted?.setVersion(initialVersion); + await worker.start(); + + const state = await client.waitForOrchestrationCompletion(id, true, 10); + + expect(state).toBeDefined(); + expect(state?.runtimeStatus).toEqual(OrchestrationStatus.COMPLETED); + expect(state?.serializedOutput).toEqual(JSON.stringify("2.0.0")); + expect(observedVersions).toEqual(["1.0.0", "2.0.0", "2.0.0"]); + }); + + it("should clear the current version when continue-as-new specifies an empty version", async () => { + const observedVersions: string[] = []; + const orchestrator: TOrchestrator = async (ctx: OrchestrationContext, input: number) => { + if (!ctx.isReplaying) { + observedVersions.push(ctx.version); + } + + if (input === 0) { + ctx.continueAsNew(1, false, ""); + return; + } + + return ctx.version; + }; + + worker.addOrchestrator(orchestrator); + const id = "clear-version-continue-as-new"; + backend.createInstance(id, getName(orchestrator), JSON.stringify(0)); + const initialExecutionStarted = backend + .getInstance(id) + ?.pendingEvents.find((event) => event.hasExecutionstarted()) + ?.getExecutionstarted(); + const initialVersion = new StringValue(); + initialVersion.setValue("2.0.0"); + initialExecutionStarted?.setVersion(initialVersion); + await worker.start(); + + const state = await client.waitForOrchestrationCompletion(id, true, 10); + + expect(state).toBeDefined(); + expect(state?.runtimeStatus).toEqual(OrchestrationStatus.COMPLETED); + expect(state?.serializedOutput).toEqual(JSON.stringify("")); + expect(observedVersions).toEqual(["2.0.0", ""]); + }); + it("should not collide default sub-orchestration instance IDs across continue-as-new generations", async () => { // Regression for the callHttp-on-continueAsNew collision: a default (auto-derived) child // instance ID must be unique per generation. Before the fix the derived ID was diff --git a/packages/durabletask-js/test/orchestration_executor.spec.ts b/packages/durabletask-js/test/orchestration_executor.spec.ts index 94a9780a..82f720a1 100644 --- a/packages/durabletask-js/test/orchestration_executor.spec.ts +++ b/packages/durabletask-js/test/orchestration_executor.spec.ts @@ -1010,6 +1010,7 @@ describe("Orchestration Executor", () => { ); expect(completeAction?.getResult()?.getValue()).toEqual(JSON.stringify(2)); expect(completeAction?.getCarryovereventsList()?.length).toEqual(saveEvent ? 3 : 0); + expect(completeAction?.getNewversion()).toBeUndefined(); for (let i = 0; i < (completeAction?.getCarryovereventsList()?.length ?? 0); i++) { const event = completeAction?.getCarryovereventsList()[i]; @@ -1024,6 +1025,29 @@ describe("Orchestration Executor", () => { } }); + it("should set the new version on a continue-as-new action", async () => { + const orchestrator: TOrchestrator = async (ctx: OrchestrationContext, input: number) => { + ctx.continueAsNew(input + 1, false, "2.0.0"); + }; + + const registry = new Registry(); + const orchestratorName = registry.addOrchestrator(orchestrator); + const newEvents = [ + newOrchestratorStartedEvent(), + newExecutionStartedEvent(orchestratorName, TEST_INSTANCE_ID, "1"), + ]; + + const executor = new OrchestrationExecutor(registry, testLogger); + const result = await executor.execute(TEST_INSTANCE_ID, [], newEvents); + + const completeAction = getAndValidateSingleCompleteOrchestrationAction(result); + expect(completeAction?.getOrchestrationstatus()).toEqual( + pb.OrchestrationStatus.ORCHESTRATION_STATUS_CONTINUED_AS_NEW, + ); + expect(completeAction?.getResult()?.getValue()).toEqual(JSON.stringify(2)); + expect(completeAction?.getNewversion()?.getValue()).toEqual("2.0.0"); + }); + it("should test that a fan-out pattern correctly schedules N tasks", async () => { const hello = async (_: any, name: string) => { return `Hello ${name}`;