diff --git a/README.md b/README.md index ae80647..6385dd6 100644 --- a/README.md +++ b/README.md @@ -5,7 +5,23 @@ This repo contains a JavaScript/TypeScript SDK for use with the [Azure Durable Task Scheduler](https://github.com/Azure/Durable-Task-Scheduler). With this SDK, you can define, schedule, and manage durable orchestrations using ordinary TypeScript/JavaScript code. -> Note that this SDK is **not** currently compatible with [Azure Durable Functions](https://learn.microsoft.com/azure/azure-functions/durable/durable-functions-overview). If you are looking for a JavaScript SDK for Azure Durable Functions, please see [this repo](https://github.com/Azure/azure-functions-durable-js). +> Note that this SDK does **not** provide the [Azure Durable Functions](https://learn.microsoft.com/azure/azure-functions/durable/durable-functions-overview) programming model, decorators, or worker-indexing metadata. If you are looking for a JavaScript SDK for Azure Durable Functions, please see [this repo](https://github.com/Azure/azure-functions-durable-js). This package exposes low-level TaskHubSidecarService gRPC/protobuf helpers that host integrations can reuse; those helpers follow this package's Node.js 22+ requirement. + +## Low-level host integration APIs + +Host integrations that already own trigger metadata and transport encoding can depend on the `@microsoft/durabletask-js` package directly. `TaskHubGrpcWorker` registers orchestrators, activities, and entities, and can process raw TaskHubSidecarService protobuf payloads without starting the long-running gRPC worker loop: + +```typescript +const worker = new TaskHubGrpcWorker(); +worker.addOrchestrator(myOrchestrator); +worker.addActivity(myActivity); +worker.addEntity(myEntity); + +const orchestrationResponseBytes = await worker.processOrchestratorRequest(orchestrationRequestBytes); +const entityResponseBytes = await worker.processEntityBatchRequest(entityBatchRequestBytes); +``` + +`TaskHubGrpcClient` already exposes orchestration start/query/event/terminate/suspend/resume/purge APIs and entity signal/read/query/clean APIs through its existing `hostAddress` and `metadataGenerator` options. Host integrations that need task-hub routing metadata should provide it through `metadataGenerator`, keeping host-specific metadata policy outside the core client. Azure-managed scheduler connection strings remain in `@microsoft/durabletask-js-azuremanaged`. ## npm packages diff --git a/eng/ci/release.yml b/eng/ci/release.yml index 0ac44d8..084871d 100644 --- a/eng/ci/release.yml +++ b/eng/ci/release.yml @@ -84,3 +84,34 @@ extends: serviceendpointurl: 'https://api.esrp.microsoft.com' mainpublisher: 'durabletask-java' domaintenantid: '33e01921-4d64-4f8c-a055-5bdaffd5e33d' + + - job: durable_functions + displayName: 'Release durable-functions' + templateContext: + type: releaseJob + isProduction: true + inputs: + - input: pipelineArtifact + pipeline: DurableTaskJSBuildPipeline + artifactName: drop + targetPath: $(System.DefaultWorkingDirectory)/drop + + steps: + - task: SFP.release-tasks.custom-build-release-task.EsrpRelease@9 + displayName: 'ESRP Release durable-functions' + inputs: + connectedservicename: 'dtfx-internal-esrp-prod' + usemanagedidentity: true + keyvaultname: 'durable-esrp-akv' + signcertname: 'dts-esrp-cert' + clientid: '0b3ed1a4-0727-4a50-b82a-02c2bd9dec89' + intent: 'PackageDistribution' + contenttype: 'npm' + contentsource: 'Folder' + folderlocation: '$(System.DefaultWorkingDirectory)/drop/buildoutputs/azure-functions-durable' + waitforreleasecompletion: true + owners: 'wangbill@microsoft.com' + approvers: 'kaibocai@microsoft.com' + serviceendpointurl: 'https://api.esrp.microsoft.com' + mainpublisher: 'durabletask-java' + domaintenantid: '33e01921-4d64-4f8c-a055-5bdaffd5e33d' diff --git a/eng/templates/build.yml b/eng/templates/build.yml index b59cdb0..0d69813 100644 --- a/eng/templates/build.yml +++ b/eng/templates/build.yml @@ -44,6 +44,12 @@ jobs: npm pack displayName: "pack @microsoft/durabletask-js-azuremanaged" + # Pack durable-functions + - script: | + cd packages/azure-functions-durable + npm pack + displayName: "pack durable-functions" + # Copy each package to its own staging subfolder - task: CopyFiles@2 displayName: "Copy durabletask-js to staging" @@ -56,4 +62,10 @@ jobs: inputs: SourceFolder: $(System.DefaultWorkingDirectory)/packages/durabletask-js-azuremanaged Contents: "*.tgz" - TargetFolder: $(Build.ArtifactStagingDirectory)/buildoutputs/durabletask-js-azuremanaged \ No newline at end of file + TargetFolder: $(Build.ArtifactStagingDirectory)/buildoutputs/durabletask-js-azuremanaged + - task: CopyFiles@2 + displayName: "Copy azure-functions-durable to staging" + inputs: + SourceFolder: $(System.DefaultWorkingDirectory)/packages/azure-functions-durable + Contents: "*.tgz" + TargetFolder: $(Build.ArtifactStagingDirectory)/buildoutputs/azure-functions-durable \ No newline at end of file diff --git a/package-lock.json b/package-lock.json index edf9cdb..f9c04e0 100644 --- a/package-lock.json +++ b/package-lock.json @@ -182,6 +182,28 @@ "node": ">=20.0.0" } }, + "node_modules/@azure/functions": { + "version": "4.16.1", + "resolved": "https://registry.npmjs.org/@azure/functions/-/functions-4.16.1.tgz", + "integrity": "sha512-A9obwC7IBg4NAmxUfTVfYEd8Xg6Px+o85JRprS3UJZt+GYYzIOmEecnFwTe3rl+aiHDewBk/8fnIVrSjR/fNGQ==", + "license": "MIT", + "dependencies": { + "@azure/functions-extensions-base": "0.3.0", + "cookie": "^0.7.0" + }, + "engines": { + "node": ">=20.0" + } + }, + "node_modules/@azure/functions-extensions-base": { + "version": "0.3.0", + "resolved": "https://registry.npmjs.org/@azure/functions-extensions-base/-/functions-extensions-base-0.3.0.tgz", + "integrity": "sha512-Cux0hLu5ZXlC/Kb+yvJVhRLIdkfFwui2HeT5oGZL00r/GCUUkhGTzRfZUjRN4Bq729mPv3okPucz2z7SMQLStA==", + "license": "MIT", + "engines": { + "node": ">=18.0" + } + }, "node_modules/@azure/identity": { "version": "4.13.1", "resolved": "https://registry.npmjs.org/@azure/identity/-/identity-4.13.1.tgz", @@ -3182,6 +3204,15 @@ "dev": true, "license": "MIT" }, + "node_modules/cookie": { + "version": "0.7.2", + "resolved": "https://registry.npmjs.org/cookie/-/cookie-0.7.2.tgz", + "integrity": "sha512-yki5XnKuf750l50uGTllt6kKILY4nQ1eNIQatoXEByZ5dWgnKqbnqmTrBE5B4N7lrMJKQ2ytWMiTO2o0v6Ew/w==", + "license": "MIT", + "engines": { + "node": ">= 0.6" + } + }, "node_modules/create-jest": { "version": "29.7.0", "resolved": "https://registry.npmjs.org/create-jest/-/create-jest-29.7.0.tgz", @@ -3368,6 +3399,10 @@ "url": "https://dotenvx.com" } }, + "node_modules/durable-functions": { + "resolved": "packages/azure-functions-durable", + "link": true + }, "node_modules/ecdsa-sig-formatter": { "version": "1.0.11", "resolved": "https://registry.npmjs.org/ecdsa-sig-formatter/-/ecdsa-sig-formatter-1.0.11.tgz", @@ -7534,6 +7569,43 @@ "url": "https://github.com/sponsors/sindresorhus" } }, + "packages/azure-functions-durable": { + "name": "durable-functions", + "version": "4.0.0-alpha.0", + "license": "MIT", + "dependencies": { + "@azure/functions": "^4.16.1", + "@grpc/grpc-js": "^1.14.4", + "@microsoft/durabletask-js": "0.3.0" + }, + "devDependencies": { + "@types/jest": "^29.5.1", + "@types/node": "^18.16.1", + "jest": "^29.5.0", + "ts-jest": "^29.1.0", + "typescript": "^5.0.4" + }, + "engines": { + "node": ">=22.0.0" + } + }, + "packages/azure-functions-durable/node_modules/@types/node": { + "version": "18.19.130", + "resolved": "https://registry.npmjs.org/@types/node/-/node-18.19.130.tgz", + "integrity": "sha512-GRaXQx6jGfL8sKfaIDD6OupbIHBr9jv7Jnaml9tB7l4v068PAOXqfcujMMo5PhbIs6ggR1XODELqahT2R8v0fg==", + "dev": true, + "license": "MIT", + "dependencies": { + "undici-types": "~5.26.4" + } + }, + "packages/azure-functions-durable/node_modules/undici-types": { + "version": "5.26.5", + "resolved": "https://registry.npmjs.org/undici-types/-/undici-types-5.26.5.tgz", + "integrity": "sha512-JlCMO+ehdEIKqlFxk6IfVoAUVmgz7cU7zD/h9XZ0qzeosSHmUJVOzSQvvYSYWXkFXC+IfLKSIffhv0sVZup6pA==", + "dev": true, + "license": "MIT" + }, "packages/durabletask-js": { "name": "@microsoft/durabletask-js", "version": "0.3.0", diff --git a/packages/azure-functions-durable/CHANGELOG.md b/packages/azure-functions-durable/CHANGELOG.md new file mode 100644 index 0000000..b2c1a44 --- /dev/null +++ b/packages/azure-functions-durable/CHANGELOG.md @@ -0,0 +1,5 @@ +# Changelog + +## 4.0.0-alpha.0 + +- Added the initial gRPC-consolidated Azure Functions Durable provider package. diff --git a/packages/azure-functions-durable/LICENSE b/packages/azure-functions-durable/LICENSE new file mode 100644 index 0000000..22aed37 --- /dev/null +++ b/packages/azure-functions-durable/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) Microsoft Corporation. + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/packages/azure-functions-durable/README.md b/packages/azure-functions-durable/README.md new file mode 100644 index 0000000..7dfd1ce --- /dev/null +++ b/packages/azure-functions-durable/README.md @@ -0,0 +1,50 @@ +# durable-functions + +Write [Azure Durable Functions](https://learn.microsoft.com/azure/azure-functions/durable/) orchestrations, activities, and entities in JavaScript and TypeScript, running on the [Durable Task JavaScript SDK](https://github.com/microsoft/durabletask-js). + +## What this package is + +`durable-functions` is the Azure Functions Durable provider for JavaScript, built on `@microsoft/durabletask-js`. You author Durable Functions apps with the familiar `app.orchestration` / `app.activity` / `app.entity` model, and the provider talks to the Durable Task backend over the Functions host's gRPC channel. + +This is a new major version that supersedes the legacy [`durable-functions`](https://github.com/Azure/azure-functions-durable-js) package. New and existing (classic v3) orchestration, activity, and entity code all run on the same gRPC engine. + +## Why it is needed + +- **One gRPC protocol.** Durable work items flow between the Functions host and your app over a single gRPC channel instead of the legacy out-of-proc HTTP protocol, keeping the JavaScript provider aligned with the .NET, Python, and Java Durable providers. +- **Shared core engine.** Orchestration replay, activities, entities, retries, and instance-management APIs all come from `@microsoft/durabletask-js`, so behavior matches the other Durable Task SDKs. +- **Backward compatible.** Existing v3 Durable Functions orchestrators and entities keep working through a compatibility layer, so you can move to this provider without rewriting your functions. + +## What it supports + +- **Authoring** — `app.orchestration`, `app.activity`, and `app.entity` register durable functions (each trigger opts into the host's gRPC protocol automatically). +- **Client** — `getClient(context)` returns a `DurableFunctionsClient` for scheduling, querying, signaling, and managing instances, plus HTTP management-payload helpers (`createCheckStatusResponse`, `createHttpManagementPayload`) for durable HTTP starters. +- **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. + +## Getting started + +```typescript +import * as df from "durable-functions"; +import { app, HttpRequest, HttpResponseInit, InvocationContext } from "@azure/functions"; + +df.app.orchestration("helloOrchestrator", async function* (ctx, input) { + return yield ctx.callActivity("sayHello", input); +}); + +df.app.activity("sayHello", { + handler: (name: string) => `Hello, ${name}!`, +}); + +app.http("startHello", { + route: "orchestrators/helloOrchestrator", + extraInputs: [df.input.durableClient()], + handler: async (request: HttpRequest, context: InvocationContext): Promise => { + const client = df.getClient(context); + const instanceId = await client.scheduleNewOrchestration("helloOrchestrator", "Durable"); + return client.createCheckStatusResponse(request, instanceId); + }, +}); +``` + +## Status + +This package is an early preview (`4.0.0-alpha.0`); APIs may change before the stable release. \ No newline at end of file diff --git a/packages/azure-functions-durable/jest.config.js b/packages/azure-functions-durable/jest.config.js new file mode 100644 index 0000000..a830572 --- /dev/null +++ b/packages/azure-functions-durable/jest.config.js @@ -0,0 +1,12 @@ +module.exports = { + preset: "ts-jest", + testEnvironment: "node", + testMatch: ["**/test/**/*.spec.ts"], + moduleFileExtensions: ["ts", "tsx", "js", "jsx", "json", "node"], + transform: { + "^.+\\.ts$": "ts-jest", + }, + moduleNameMapper: { + "^@microsoft/durabletask-js$": "/../durabletask-js/src/index.ts", + }, +}; diff --git a/packages/azure-functions-durable/package.json b/packages/azure-functions-durable/package.json new file mode 100644 index 0000000..128ca9f --- /dev/null +++ b/packages/azure-functions-durable/package.json @@ -0,0 +1,63 @@ +{ + "name": "durable-functions", + "version": "4.0.0-alpha.0", + "description": "Azure Functions Durable provider for the Durable Task JavaScript SDK", + "main": "./dist/index.js", + "types": "./dist/index.d.ts", + "exports": { + ".": { + "types": "./dist/index.d.ts", + "require": "./dist/index.js", + "import": "./dist/index.js" + } + }, + "files": [ + "dist", + "LICENSE", + "README.md", + "CHANGELOG.md" + ], + "scripts": { + "clean": "node -e \"require('fs').rmSync('dist', {recursive:true, force:true})\"", + "prebuild": "node -p \"'// Auto-generated by prebuild\\nexport const SDK_VERSION = ' + JSON.stringify(require('./package.json').version) + ';\\nexport const SDK_PACKAGE_NAME = ' + JSON.stringify(require('./package.json').name) + ';'\" > src/version.ts", + "build:core": "npm run build -w @microsoft/durabletask-js", + "build": "npm run prebuild && npm run clean && npm run build:core && tsc -p tsconfig.build.json", + "test": "jest --runInBand --detectOpenHandles", + "test:unit": "jest test/unit --runInBand --detectOpenHandles", + "prepublishOnly": "npm run build && npm run test" + }, + "repository": { + "type": "git", + "url": "git+https://github.com/microsoft/durabletask-js.git", + "directory": "packages/azure-functions-durable" + }, + "keywords": [ + "azure-functions", + "durable-functions", + "durabletask", + "orchestration", + "workflow", + "grpc" + ], + "author": "Microsoft", + "license": "MIT", + "bugs": { + "url": "https://github.com/microsoft/durabletask-js/issues" + }, + "homepage": "https://github.com/microsoft/durabletask-js/tree/main/packages/azure-functions-durable#readme", + "engines": { + "node": ">=22.0.0" + }, + "dependencies": { + "@azure/functions": "^4.16.1", + "@grpc/grpc-js": "^1.14.4", + "@microsoft/durabletask-js": "0.3.0" + }, + "devDependencies": { + "@types/jest": "^29.5.1", + "@types/node": "^18.16.1", + "jest": "^29.5.0", + "ts-jest": "^29.1.0", + "typescript": "^5.0.4" + } +} diff --git a/packages/azure-functions-durable/src/app.ts b/packages/azure-functions-durable/src/app.ts new file mode 100644 index 0000000..c8e0040 --- /dev/null +++ b/packages/azure-functions-durable/src/app.ts @@ -0,0 +1,154 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { + FunctionHandler, + FunctionInput, + FunctionOutput, + InvocationContext, + app as azFuncApp, +} from "@azure/functions"; +import { EntityFactory, TOrchestrator } from "@microsoft/durabletask-js"; +import * as trigger from "./trigger"; +import { ClassicEntity, wrapEntity } from "./entity-context"; +import { ClassicOrchestrator, wrapOrchestrator } from "./orchestration-context"; +import { DurableFunctionsWorker } from "./worker"; + +// A single worker owns the orchestrator/entity registry for the whole function app. It is never +// started (no gRPC channel, no work-item loop); it only replays one work item per invocation via +// the byte-processor methods. Orchestrators and entities are pre-registered at decoration time so +// the core executor can dispatch by name when the host delivers a work item. This mirrors Python's +// provider intent while avoiding a new worker (and re-registration) per invocation. +const sharedWorker = new DurableFunctionsWorker(); + +/** + * Returns the app-wide {@link DurableFunctionsWorker} that holds the orchestrator/entity registry. + * Exposed for host integration and testing; application code does not normally need it. + */ +export function getSharedWorker(): DurableFunctionsWorker { + return sharedWorker; +} + +/** Secondary bindings that pass straight through to `@azure/functions` `app.generic`. */ +interface ExtraRegistrationOptions { + extraInputs?: FunctionInput[]; + extraOutputs?: FunctionOutput[]; +} + +// Handlers accept both the core-native form and the classic Durable Functions (v3) form; the +// wrappers below detect which was provided and adapt the classic form onto the core engine. +export type OrchestrationHandler = TOrchestrator | ClassicOrchestrator; +export type OrchestrationOptions = ExtraRegistrationOptions & { handler: OrchestrationHandler }; + +export type EntityHandler = EntityFactory | ClassicEntity; +export type EntityOptions = ExtraRegistrationOptions & { handler: EntityHandler }; + +export type ActivityHandler = FunctionHandler; +export type ActivityOptions = ExtraRegistrationOptions & { handler: ActivityHandler }; + +/** + * Registers an orchestrator function. The generator is registered on the shared worker under + * `functionName`, and the Azure Function is wired to the gRPC byte-processing handler so the host's + * base64 `OrchestratorRequest` is replayed by the core executor. + */ +export function orchestration( + functionName: string, + handlerOrOptions: OrchestrationHandler | OrchestrationOptions, +): void { + const options = normalizeOptions(handlerOrOptions); + sharedWorker.addNamedOrchestrator(functionName, wrapOrchestrator(options.handler)); + azFuncApp.generic(functionName, { + ...extraBindings(options), + trigger: trigger.orchestration(), + handler: createOrchestrationHandler(), + }); +} + +/** + * Registers an entity function. The entity factory is registered on the shared worker under + * `functionName`, and the Azure Function is wired to the gRPC byte-processing handler so the host's + * base64 `EntityBatchRequest` is replayed by the core executor. + */ +export function entity(functionName: string, handlerOrOptions: EntityHandler | EntityOptions): void { + const options = normalizeOptions(handlerOrOptions); + sharedWorker.addNamedEntity(functionName, wrapEntity(options.handler)); + azFuncApp.generic(functionName, { + ...extraBindings(options), + trigger: trigger.entity(), + handler: createEntityHandler(), + }); +} + +/** + * Registers an activity function. Activities are plain Azure Functions: the host delivers the input + * as the trigger payload and takes the return value directly, so the user handler is registered + * as-is (no worker round-trip). The `activityTrigger` still carries `durableRequiresGrpc: true` so + * the extension keeps the gRPC protocol enabled for the app. + */ +export function activity( + functionName: string, + handlerOrOptions: ActivityHandler | ActivityOptions, +): void { + const options = normalizeOptions(handlerOrOptions); + azFuncApp.generic(functionName, { + ...extraBindings(options), + trigger: trigger.activity(), + handler: options.handler, + }); +} + +/** @hidden */ +function createOrchestrationHandler(): FunctionHandler { + return async (triggerInput: unknown, _context: InvocationContext): Promise => { + return sharedWorker.handleOrchestratorRequest(extractBase64Request(triggerInput)); + }; +} + +/** @hidden */ +function createEntityHandler(): FunctionHandler { + return async (triggerInput: unknown, _context: InvocationContext): Promise => { + return sharedWorker.handleEntityBatchRequest(extractBase64Request(triggerInput)); + }; +} + +/** @hidden */ +function normalizeOptions( + handlerOrOptions: THandler | (ExtraRegistrationOptions & { handler: THandler }), +): ExtraRegistrationOptions & { handler: THandler } { + if (typeof handlerOrOptions === "function") { + return { handler: handlerOrOptions as THandler }; + } + return handlerOrOptions as ExtraRegistrationOptions & { handler: THandler }; +} + +/** @hidden */ +function extraBindings(options: ExtraRegistrationOptions): ExtraRegistrationOptions { + const bindings: ExtraRegistrationOptions = {}; + if (options.extraInputs) { + bindings.extraInputs = options.extraInputs; + } + if (options.extraOutputs) { + bindings.extraOutputs = options.extraOutputs; + } + return bindings; +} + +/** + * Extracts the base64-encoded protobuf request the host delivers to a durable trigger. The value + * may arrive as the raw base64 string, or wrapped in an object exposing a string `body` (mirrors + * the Python worker's `getattr(context, "body")` fallback to the context itself). + * + * @hidden + */ +function extractBase64Request(triggerInput: unknown): string { + if (typeof triggerInput === "string") { + return triggerInput; + } + if (triggerInput !== null && typeof triggerInput === "object" && "body" in triggerInput) { + const body = (triggerInput as { body?: unknown }).body; + if (typeof body === "string") { + return body; + } + } + throw new TypeError("Durable trigger did not provide a base64-encoded request body."); +} diff --git a/packages/azure-functions-durable/src/client.ts b/packages/azure-functions-durable/src/client.ts new file mode 100644 index 0000000..804d785 --- /dev/null +++ b/packages/azure-functions-durable/src/client.ts @@ -0,0 +1,502 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { HttpRequest, HttpResponse } from "@azure/functions"; +import { + EntityInstanceId, + OrchestrationQuery, + OrchestrationStatus, + PurgeInstanceCriteria, + TaskHubGrpcClient, +} from "@microsoft/durabletask-js"; +import { + HttpManagementPayload, + createHttpManagementPayload as createPayload, +} from "./http-management-payload"; +import { EntityStateResponse } from "./entity-state-response"; +import { createAzureFunctionsMetadataGenerator } from "./metadata"; +import { + DurableOrchestrationStatus, + fromOrchestrationRuntimeStatus, + toDurableOrchestrationStatus, +} from "./orchestration-status"; +import { OrchestrationFilter } from "./orchestration-filter"; +import { PurgeHistoryResult } from "./purge-history-result"; + +export interface DurableFunctionsClientConfig { + taskHubName?: string; + connectionName?: string; + creationUrls?: Record; + managementUrls?: Record; + baseUrl?: string; + requiredQueryStringParameters?: string; + rpcBaseUrl?: string; + httpBaseUrl?: string; + maxGrpcMessageSizeInBytes?: number; + grpcHttpClientTimeout?: unknown; +} + +export type DurableFunctionsClientInput = string | DurableFunctionsClientConfig; + +export class DurableFunctionsClient extends TaskHubGrpcClient { + public readonly taskHubName: string; + public readonly connectionName: string; + public readonly creationUrls: Record; + public readonly managementUrls: Record; + public readonly baseUrl: string; + public readonly requiredQueryStringParameters: string; + public readonly rpcBaseUrl: string; + public readonly httpBaseUrl: string; + public readonly maxGrpcMessageSizeInBytes: number; + public readonly grpcHttpClientTimeout: unknown; + + constructor(clientConfig: DurableFunctionsClientInput) { + const config = parseClientConfig(clientConfig); + const taskHubName = config.taskHubName ?? ""; + const requiredQueryStringParameters = config.requiredQueryStringParameters ?? ""; + const rpcBaseUrl = requireString(config.rpcBaseUrl, "rpcBaseUrl"); + const maxGrpcMessageSizeInBytes = config.maxGrpcMessageSizeInBytes ?? 0; + + super({ + hostAddress: getGrpcHostAddress(rpcBaseUrl), + useTLS: false, + metadataGenerator: createAzureFunctionsMetadataGenerator(taskHubName), + // Honor the host-provided gRPC message size limit, matching the Python provider. When unset + // (0), the gRPC library defaults are left in place. + options: + maxGrpcMessageSizeInBytes > 0 + ? { + "grpc.max_receive_message_length": maxGrpcMessageSizeInBytes, + "grpc.max_send_message_length": maxGrpcMessageSizeInBytes, + } + : undefined, + }); + + this.taskHubName = taskHubName; + this.connectionName = config.connectionName ?? ""; + this.creationUrls = config.creationUrls ?? {}; + this.managementUrls = config.managementUrls ?? {}; + this.baseUrl = config.baseUrl ?? ""; + this.requiredQueryStringParameters = requiredQueryStringParameters; + this.rpcBaseUrl = rpcBaseUrl; + this.httpBaseUrl = config.httpBaseUrl ?? ""; + this.maxGrpcMessageSizeInBytes = maxGrpcMessageSizeInBytes; + this.grpcHttpClientTimeout = config.grpcHttpClientTimeout; + } + + createCheckStatusResponse(request: HttpRequest, instanceId: string): HttpResponse { + const payload = this.createHttpManagementPayload(request, instanceId); + + return new HttpResponse({ + status: 202, + body: JSON.stringify(payload), + headers: { + "content-type": "application/json", + Location: payload.statusQueryGetUri, + }, + }); + } + + /** + * Builds the {@link HttpManagementPayload} of management URLs for an orchestration instance. + * + * @remarks + * Two call styles are supported (mirroring the Python provider's backward-compatible surface): + * - `createHttpManagementPayload(request, instanceId)` (recommended): builds the URLs relative to + * the incoming request's origin. + * - `createHttpManagementPayload(instanceId)` (classic Durable Functions v3 style): builds the + * URLs from the client binding's `baseUrl` when no request is available. + */ + createHttpManagementPayload(instanceId: string): HttpManagementPayload; + createHttpManagementPayload(request: HttpRequest, instanceId: string): HttpManagementPayload; + createHttpManagementPayload( + requestOrInstanceId: HttpRequest | string, + instanceId?: string, + ): HttpManagementPayload { + // Classic Durable Functions (v3) accepted a single positional `instanceId`. Detect that call + // style (a lone string argument) and fall back to the client binding's `baseUrl` when building + // the payload URLs. + let request: HttpRequest | undefined; + if (typeof requestOrInstanceId === "string") { + instanceId = requestOrInstanceId; + } else { + request = requestOrInstanceId; + } + if (instanceId === undefined) { + throw new TypeError("instanceId is required."); + } + const instanceStatusUrl = getInstanceStatusUrl(request, instanceId, this.baseUrl); + return createPayload(instanceId, instanceStatusUrl, this.requiredQueryStringParameters); + } + + /** + * Deprecated alias for {@link createHttpManagementPayload} (classic Durable Functions v3 shape). + * + * @deprecated Use {@link createHttpManagementPayload} instead. + * @param request - The incoming HTTP request, or `undefined` to build the URLs from the client + * binding's `baseUrl`. + * @param instanceId - The orchestration instance to build management URLs for. + */ + getClientResponseLinks( + request: HttpRequest | undefined, + instanceId: string, + ): HttpManagementPayload { + return request === undefined + ? this.createHttpManagementPayload(instanceId) + : this.createHttpManagementPayload(request, instanceId); + } + + /** + * Starts a new orchestration instance (classic Durable Functions v3 `startNew` alias). + * + * @deprecated Use {@link scheduleNewOrchestration} instead. + * @param orchestratorName - The name of the orchestrator to start. + * @param options - Optional input and instance ID. + * @returns The instance ID of the started orchestration. + */ + async startNew( + orchestratorName: string, + options?: { input?: unknown; instanceId?: string }, + ): Promise { + return this.scheduleNewOrchestration( + orchestratorName, + options?.input, + options?.instanceId !== undefined ? { instanceId: options.instanceId } : undefined, + ); + } + + /** + * Gets the status of an orchestration instance in the classic Durable Functions (v3) shape. + * + * @deprecated Use {@link getOrchestrationState} instead. + * @param instanceId - The ID of the orchestration instance to query. + * @param options - When `showInput` is `false`, input/output payloads are not fetched. + * @returns The instance status, or `undefined` if the instance does not exist. + */ + async getStatus( + instanceId: string, + options?: { showInput?: boolean }, + ): Promise { + const state = await this.getOrchestrationState(instanceId, options?.showInput ?? true); + return state ? toDurableOrchestrationStatus(state) : undefined; + } + + /** + * Gets the status of all orchestration instances (classic Durable Functions v3 shape). + * + * @deprecated Use {@link getAllInstances} instead. + */ + async getStatusAll(): Promise { + return this.collectStatuses({ fetchInputsAndOutputs: true }); + } + + /** + * Gets the status of orchestration instances matching a filter (classic Durable Functions v3 shape). + * + * @deprecated Use {@link getAllInstances} instead. + * @param filter - Creation-time window and/or runtime-status filter. + */ + async getStatusBy(filter: OrchestrationFilter): Promise { + return this.collectStatuses({ + createdFrom: filter.createdTimeFrom, + createdTo: filter.createdTimeTo, + statuses: filter.runtimeStatus?.map(fromOrchestrationRuntimeStatus), + fetchInputsAndOutputs: true, + }); + } + + /** + * Waits up to a timeout for an orchestration to complete; if it does, returns an HTTP response with + * its output/status, otherwise returns the same 202 check-status response as + * {@link createCheckStatusResponse}. Classic Durable Functions v3 behavior. + * + * @deprecated Use {@link waitForOrchestrationCompletion} together with + * {@link createCheckStatusResponse} instead. + * @param request - The incoming HTTP request (used to build management URLs on timeout). + * @param instanceId - The orchestration instance to wait for. + * @param waitOptions - Optional total wait timeout in milliseconds (default 10s). + */ + async waitForCompletionOrCreateCheckStatusResponse( + request: HttpRequest, + instanceId: string, + waitOptions?: { timeoutInMilliseconds?: number }, + ): Promise { + const timeoutSeconds = Math.max( + 1, + Math.ceil((waitOptions?.timeoutInMilliseconds ?? 10000) / 1000), + ); + try { + const state = await this.waitForOrchestrationCompletion(instanceId, true, timeoutSeconds); + if (state) { + const headers = { "content-type": "application/json" }; + if (state.runtimeStatus === OrchestrationStatus.COMPLETED) { + return new HttpResponse({ status: 200, body: state.serializedOutput ?? "null", headers }); + } + if (state.runtimeStatus === OrchestrationStatus.FAILED) { + return new HttpResponse({ + status: 500, + body: JSON.stringify(toDurableOrchestrationStatus(state)), + headers, + }); + } + if (state.runtimeStatus === OrchestrationStatus.TERMINATED) { + return new HttpResponse({ + status: 200, + body: JSON.stringify(toDurableOrchestrationStatus(state)), + headers, + }); + } + } + } catch { + // Timed out (or not yet terminal) waiting for completion: fall through to the check-status + // response so the caller can poll the management endpoints. + } + return this.createCheckStatusResponse(request, instanceId); + } + + /** + * Reads the state of a durable entity in the classic Durable Functions (v3) shape. + * + * @deprecated Use {@link getEntity} instead. + * @param entityId - The target entity instance ID. + * @param includeState - Whether to include the entity state in the response (default `true`). + */ + async readEntityState( + entityId: EntityInstanceId, + includeState = true, + ): Promise> { + const metadata = await this.getEntity(entityId, includeState); + if (!metadata) { + return new EntityStateResponse(false); + } + return new EntityStateResponse(true, metadata.state); + } + + /** + * Purges the history of a single orchestration instance, returning the classic Durable Functions + * (v3) {@link PurgeHistoryResult}. + * + * @deprecated Use {@link purgeOrchestration} instead. + * @param instanceId - The ID of the orchestration instance to purge. + */ + async purgeInstanceHistory(instanceId: string): Promise { + const result = await this.purgeOrchestration(instanceId); + return new PurgeHistoryResult(result?.deletedInstanceCount ?? 0); + } + + /** + * Purges the history of orchestration instances matching a filter (classic Durable Functions v3 + * shape), returning a {@link PurgeHistoryResult}. + * + * @deprecated Use {@link purgeOrchestration} with a {@link PurgeInstanceCriteria} instead. + * @param filter - Creation-time window and/or runtime-status filter. + */ + async purgeInstanceHistoryBy(filter: OrchestrationFilter): Promise { + const criteria = new PurgeInstanceCriteria(); + criteria.setCreatedTimeFrom(filter.createdTimeFrom); + criteria.setCreatedTimeTo(filter.createdTimeTo); + if (filter.runtimeStatus) { + criteria.setRuntimeStatusList(filter.runtimeStatus.map(fromOrchestrationRuntimeStatus)); + } + const result = await this.purgeOrchestration(criteria); + return new PurgeHistoryResult(result?.deletedInstanceCount ?? 0); + } + + /** + * Raises an external event to an orchestration instance (classic Durable Functions v3 alias). + * + * @deprecated Use {@link raiseOrchestrationEvent} instead. + * @param instanceId - The orchestration instance to raise the event on. + * @param eventName - The name of the event (case-insensitive). + * @param eventData - Optional JSON-serializable event payload. + */ + async raiseEvent(instanceId: string, eventName: string, eventData?: unknown): Promise { + await this.raiseOrchestrationEvent(instanceId, eventName, eventData ?? null); + } + + /** + * Terminates an orchestration instance (classic Durable Functions v3 alias). + * + * @deprecated Use {@link terminateOrchestration} instead. + * @param instanceId - The orchestration instance to terminate. + * @param reason - Optional reason recorded as the terminated instance's output. + */ + async terminate(instanceId: string, reason?: unknown): Promise { + await this.terminateOrchestration(instanceId, reason ?? null); + } + + /** + * Suspends a running orchestration instance (classic Durable Functions v3 alias). + * + * @deprecated Use {@link suspendOrchestration} instead. + * @param instanceId - The orchestration instance to suspend. + * @param _reason - Accepted for classic v3 signature compatibility; ignored (the core engine does + * not record a suspend reason). + */ + async suspend(instanceId: string, _reason?: string): Promise { + await this.suspendOrchestration(instanceId); + } + + /** + * Resumes a suspended orchestration instance (classic Durable Functions v3 alias). + * + * @deprecated Use {@link resumeOrchestration} instead. + * @param instanceId - The orchestration instance to resume. + * @param _reason - Accepted for classic v3 signature compatibility; ignored. + */ + async resume(instanceId: string, _reason?: string): Promise { + await this.resumeOrchestration(instanceId); + } + + /** + * Rewinds a failed orchestration instance so it retries from its point of failure (classic + * Durable Functions v3 alias). + * + * @remarks + * Not supported: the durabletask engine has no rewind equivalent yet. This mirrors the Python + * provider, which raises for the same reason. + * @deprecated Not supported; rewind has no durabletask equivalent yet. + * @param _instanceId - The failed orchestration instance to rewind. + * @param _reason - Optional reason describing why the instance is being rewound. + * @throws Always throws: rewind is not yet supported. + */ + async rewind(_instanceId: string, _reason?: string): Promise { + throw new Error("rewind is not yet supported by durabletask."); + } + + /** + * Restarts an orchestration instance with its original input (classic Durable Functions v3 alias). + * + * @deprecated Use {@link restartOrchestration} instead. + * @param instanceId - The orchestration instance to restart. + * @param restartWithNewInstanceId - When `true`, the restarted instance is assigned a new ID. + * @returns The instance ID of the restarted orchestration. + */ + async restart(instanceId: string, restartWithNewInstanceId = false): Promise { + return this.restartOrchestration(instanceId, restartWithNewInstanceId); + } + + /** @hidden Iterates the core paged query and maps each instance to the v3 status shape. */ + private async collectStatuses(query: OrchestrationQuery): Promise { + const results: DurableOrchestrationStatus[] = []; + for await (const state of this.getAllInstances(query)) { + results.push(toDurableOrchestrationStatus(state)); + } + return results; + } +} + +/** + * @deprecated Use {@link DurableFunctionsClient} instead. Retained as a classic Durable Functions + * (v3) alias so existing code that imports `DurableOrchestrationClient` keeps working. + */ +export class DurableOrchestrationClient extends DurableFunctionsClient {} + +export function getGrpcHostAddress(rpcBaseUrl: string): string { + try { + const hostAddress = new URL(rpcBaseUrl).host; + if (!hostAddress) { + throw new Error("rpcBaseUrl must include a host."); + } + return hostAddress; + } catch (e) { + throw new Error(`Invalid Durable Functions rpcBaseUrl: ${rpcBaseUrl}`, { cause: e }); + } +} + +function getInstanceStatusUrl( + request: HttpRequest | undefined, + instanceId: string, + baseUrl: string, +): string { + const encodedInstanceId = encodeURIComponent(instanceId); + if (request !== undefined) { + const requestUrl = new URL(request.url); + return `${requestUrl.protocol}//${requestUrl.host}/runtime/webhooks/durabletask/instances/${encodedInstanceId}`; + } + // No request (classic Durable Functions v3 single-argument call): fall back to the base URL + // supplied in the client binding configuration. + const trimmedBaseUrl = baseUrl.replace(/\/+$/, ""); + return `${trimmedBaseUrl}/instances/${encodedInstanceId}`; +} + +function parseClientConfig(clientConfig: DurableFunctionsClientInput): DurableFunctionsClientConfig { + const value: unknown = typeof clientConfig === "string" ? JSON.parse(clientConfig) : clientConfig; + const record = requireRecord(value, "Durable Functions client configuration"); + + return { + taskHubName: optionalString(record, "taskHubName"), + connectionName: optionalString(record, "connectionName"), + creationUrls: optionalStringRecord(record, "creationUrls"), + managementUrls: optionalStringRecord(record, "managementUrls"), + baseUrl: optionalString(record, "baseUrl"), + requiredQueryStringParameters: optionalString(record, "requiredQueryStringParameters"), + rpcBaseUrl: optionalString(record, "rpcBaseUrl"), + httpBaseUrl: optionalString(record, "httpBaseUrl"), + maxGrpcMessageSizeInBytes: optionalNumber(record, "maxGrpcMessageSizeInBytes"), + grpcHttpClientTimeout: record.grpcHttpClientTimeout, + }; +} + +function requireString(value: string | undefined, name: string): string { + if (!value) { + throw new TypeError(`Durable Functions client configuration is missing ${name}.`); + } + + return value; +} + +function requireRecord(value: unknown, name: string): Record { + if (value === null || typeof value !== "object" || Array.isArray(value)) { + throw new TypeError(`${name} must be a JSON object.`); + } + + return value as Record; +} + +function optionalString(record: Record, name: string): string | undefined { + const value = record[name]; + if (value === undefined || value === null) { + return undefined; + } + if (typeof value !== "string") { + throw new TypeError(`Durable Functions client configuration field ${name} must be a string.`); + } + + return value; +} + +function optionalNumber(record: Record, name: string): number | undefined { + const value = record[name]; + if (value === undefined || value === null) { + return undefined; + } + if (typeof value !== "number") { + throw new TypeError(`Durable Functions client configuration field ${name} must be a number.`); + } + + return value; +} + +function optionalStringRecord( + record: Record, + name: string, +): Record | undefined { + const value = record[name]; + if (value === undefined || value === null) { + return undefined; + } + + const valueRecord = requireRecord(value, `Durable Functions client configuration field ${name}`); + const result: Record = {}; + for (const [key, entry] of Object.entries(valueRecord)) { + if (typeof entry !== "string") { + throw new TypeError( + `Durable Functions client configuration field ${name}.${key} must be a string.`, + ); + } + result[key] = entry; + } + + return result; +} diff --git a/packages/azure-functions-durable/src/durable-grpc.ts b/packages/azure-functions-durable/src/durable-grpc.ts new file mode 100644 index 0000000..47b5058 --- /dev/null +++ b/packages/azure-functions-durable/src/durable-grpc.ts @@ -0,0 +1,13 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +export type DurableBindingMetadata = Record; + +export function addDurableGrpcMetadata( + binding: TBinding, +): TBinding & { durableRequiresGrpc: true } { + return { + ...binding, + durableRequiresGrpc: true, + }; +} diff --git a/packages/azure-functions-durable/src/entity-context.ts b/packages/azure-functions-durable/src/entity-context.ts new file mode 100644 index 0000000..a130488 --- /dev/null +++ b/packages/azure-functions-durable/src/entity-context.ts @@ -0,0 +1,147 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { + EntityFactory, + EntityInstanceId, + ITaskEntity, + TaskEntityOperation, +} from "@microsoft/durabletask-js"; + +/** + * Classic Durable Functions (v3) entity context, exposed to migrating entity functions as + * `context.df`. + * + * @remarks + * This is an adapter over the core durabletask {@link TaskEntityOperation}. It lets entity bodies + * written against the legacy `durable-functions` `context.df.*` API run unchanged on the + * gRPC/durabletask engine. + */ +export class DurableEntityContext { + private _result: unknown; + private _resultSet = false; + private readonly _isNewlyConstructed: boolean; + + constructor(private readonly _operation: TaskEntityOperation) { + this._isNewlyConstructed = !_operation.state.hasState; + } + + /** The name of the current operation. */ + get operationName(): string { + return this._operation.name; + } + + /** The entity type name (lowercased). */ + get entityName(): string { + return this._operation.context.id.name; + } + + /** The entity instance key (case-preserved). */ + get entityKey(): string { + return this._operation.context.id.key; + } + + /** The entity instance ID in `@name@key` form. */ + get instanceId(): string { + return this._operation.context.id.toString(); + } + + /** The entity instance ID as an {@link EntityInstanceId} object. */ + get entityId(): EntityInstanceId { + return this._operation.context.id; + } + + /** Whether this entity had no state prior to the current operation (freshly constructed). */ + get isNewlyConstructed(): boolean { + return this._isNewlyConstructed; + } + + /** Gets the input for the current operation. */ + getInput(): T | undefined { + return this._operation.getInput(); + } + + /** + * Gets the current entity state. + * + * @param initializer - Optional zero-argument callable providing the initial state when none exists. + */ + getState(initializer?: () => T): T | undefined { + const defaultValue = typeof initializer === "function" ? initializer() : undefined; + return this._operation.state.getState(defaultValue); + } + + /** Sets the entity state. Passing `null`/`undefined` deletes the entity. */ + setState(state: unknown): void { + this._operation.state.setState(state); + } + + /** Sets the result (return value) of the current operation. */ + return(result: unknown): void { + this._result = result; + this._resultSet = true; + } + + /** Deletes this entity after the operation completes. */ + destructOnExit(): void { + this._operation.state.setState(undefined); + } + + /** + * Signals another entity operation without waiting for a response (fire-and-forget). + * + * @param entityId - The target entity. + * @param operationName - The name of the operation to invoke. + * @param operationInput - Optional input for the operation. + */ + signalEntity( + entityId: EntityInstanceId, + operationName: string, + operationInput?: unknown, + ): void { + this._operation.context.signalEntity(entityId, operationName, operationInput); + } + + /** @hidden Returns the explicitly-set result, or the provided fallback when none was set. */ + resolveResult(fallback: unknown): unknown { + return this._resultSet ? this._result : fallback; + } +} + +/** The object passed to a classic (v3) entity function; its `df` is the durable entity context. */ +export interface ClassicEntityContext { + df: DurableEntityContext; +} + +/** + * A classic Durable Functions (v3) entity: a single-argument function that reads and mutates state + * through `context.df.*`. + */ +export type ClassicEntity = (context: ClassicEntityContext) => unknown | Promise; + +/** + * Adapts an entity handler for registration on the core worker. + * + * @remarks + * Core-native entities are zero-argument factories (`() => ITaskEntity`) and are returned + * unchanged. Classic v3 entities declare a single `context` parameter and use `context.df.*`; those + * are wrapped in a factory whose operation dispatch forwards to the core operation. + */ +export function wrapEntity(handler: EntityFactory | ClassicEntity): EntityFactory { + if (typeof handler === "function" && handler.length >= 1) { + const classic = handler as ClassicEntity; + return () => new ClassicEntityAdapter(classic); + } + return handler as EntityFactory; +} + +/** @hidden Bridges a classic v3 entity function to the core {@link ITaskEntity} contract. */ +class ClassicEntityAdapter implements ITaskEntity { + constructor(private readonly _fn: ClassicEntity) {} + + async run(operation: TaskEntityOperation): Promise { + const df = new DurableEntityContext(operation); + const returned = await Promise.resolve(this._fn({ df })); + return df.resolveResult(returned); + } +} diff --git a/packages/azure-functions-durable/src/entity-id.ts b/packages/azure-functions-durable/src/entity-id.ts new file mode 100644 index 0000000..7b9b860 --- /dev/null +++ b/packages/azure-functions-durable/src/entity-id.ts @@ -0,0 +1,15 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { EntityInstanceId } from "@microsoft/durabletask-js"; + +/** + * Classic Durable Functions (v3) entity identifier, constructed as `new EntityId(name, key)`. + * + * @remarks + * A thin subclass of the core {@link EntityInstanceId}, retained so existing `durable-functions` + * code that builds `new df.EntityId(name, key)` and passes it to `callEntity` / `signalEntity` / + * `readEntityState` keeps working. The entity name is normalized to lowercase and the key is + * preserved, matching the core behavior. + */ +export class EntityId extends EntityInstanceId {} diff --git a/packages/azure-functions-durable/src/entity-state-response.ts b/packages/azure-functions-durable/src/entity-state-response.ts new file mode 100644 index 0000000..399ffef --- /dev/null +++ b/packages/azure-functions-durable/src/entity-state-response.ts @@ -0,0 +1,19 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +/** + * The response returned by {@link DurableFunctionsClient.readEntityState}, matching the classic + * Durable Functions v3 `EntityStateResponse` shape. + * + * @typeParam T - The entity state type. + */ +export class EntityStateResponse { + /** + * @param entityExists - Whether the entity exists (has state). + * @param entityState - The entity state, present only when {@link entityExists} is true. + */ + constructor( + public readonly entityExists: boolean, + public readonly entityState?: T, + ) {} +} diff --git a/packages/azure-functions-durable/src/get-client.ts b/packages/azure-functions-durable/src/get-client.ts new file mode 100644 index 0000000..21f5cd9 --- /dev/null +++ b/packages/azure-functions-durable/src/get-client.ts @@ -0,0 +1,52 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { FunctionInput, InvocationContext } from "@azure/functions"; +import { DurableFunctionsClient, DurableFunctionsClientInput } from "./client"; + +/** + * Builds a {@link DurableFunctionsClient} from the `durableClient` input binding that the + * Azure Functions host provides for the current invocation. + * + * Mirrors the Python provider's `_add_rich_client` middleware, which constructs a + * `DurableFunctionsClient(starter)` from the raw client-binding value. The binding value is the + * host-provided client configuration (a JSON string or already-parsed object); the client + * constructor accepts either shape, so it is passed straight through. + * + * @param context - The invocation context for the current function call. + * @returns A rich Durable Functions client bound to the host's local gRPC sidecar. + * @throws If no `durableClient`/`orchestrationClient` input binding is registered, or its value + * is not a valid client configuration. + */ +export function getClient(context: InvocationContext): DurableFunctionsClient { + const clientInput = context.options.extraInputs.find(isDurableClientInput); + if (!clientInput) { + throw new Error( + "Could not find a registered durable client input binding. Check your extraInputs " + + "definition when registering your function.", + ); + } + + const bindingData: unknown = context.extraInputs.get(clientInput); + return new DurableFunctionsClient(asClientInput(bindingData)); +} + +/** @hidden */ +export function isDurableClientInput(input: FunctionInput): boolean { + return input.type === "durableClient" || input.type === "orchestrationClient"; +} + +/** @hidden */ +function asClientInput(bindingData: unknown): DurableFunctionsClientInput { + if (typeof bindingData === "string") { + return bindingData; + } + if (bindingData !== null && typeof bindingData === "object") { + return bindingData as DurableFunctionsClientInput; + } + + throw new Error( + "Received input is not a valid durable client input. Check your extraInputs definition " + + "when registering your function.", + ); +} diff --git a/packages/azure-functions-durable/src/http-management-payload.ts b/packages/azure-functions-durable/src/http-management-payload.ts new file mode 100644 index 0000000..640d7ad --- /dev/null +++ b/packages/azure-functions-durable/src/http-management-payload.ts @@ -0,0 +1,42 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +export interface HttpManagementPayload { + id: string; + purgeHistoryDeleteUri: string; + restartPostUri: string; + rewindPostUri: string; + sendEventPostUri: string; + statusQueryGetUri: string; + terminatePostUri: string; + resumePostUri: string; + suspendPostUri: string; +} + +export function createHttpManagementPayload( + instanceId: string, + instanceStatusUrl: string, + requiredQueryStringParameters: string, +): HttpManagementPayload { + const queryString = normalizeQueryString(requiredQueryStringParameters); + const querySuffix = queryString ? `?${queryString}` : ""; + const reasonQuerySuffix = queryString ? `?reason={text}&${queryString}` : "?reason={text}"; + + return { + id: instanceId, + purgeHistoryDeleteUri: `${instanceStatusUrl}${querySuffix}`, + restartPostUri: `${instanceStatusUrl}/restart${querySuffix}`, + rewindPostUri: `${instanceStatusUrl}/rewind${reasonQuerySuffix}`, + sendEventPostUri: `${instanceStatusUrl}/raiseEvent/{eventName}${querySuffix}`, + statusQueryGetUri: `${instanceStatusUrl}${querySuffix}`, + terminatePostUri: `${instanceStatusUrl}/terminate${reasonQuerySuffix}`, + resumePostUri: `${instanceStatusUrl}/resume${reasonQuerySuffix}`, + suspendPostUri: `${instanceStatusUrl}/suspend${reasonQuerySuffix}`, + }; +} + +function normalizeQueryString(requiredQueryStringParameters: string): string { + return requiredQueryStringParameters.startsWith("?") + ? requiredQueryStringParameters.slice(1) + : requiredQueryStringParameters; +} diff --git a/packages/azure-functions-durable/src/index.ts b/packages/azure-functions-durable/src/index.ts new file mode 100644 index 0000000..c7f1ad1 --- /dev/null +++ b/packages/azure-functions-durable/src/index.ts @@ -0,0 +1,66 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import type { ClassicEntityContext, ClassicEntity } from "./entity-context"; +import type { DurableFunctionsClient } from "./client"; + +export * as app from "./app"; +export * as trigger from "./trigger"; +export * as input from "./input"; + +export { + DurableFunctionsClient, + DurableFunctionsClientConfig, + DurableFunctionsClientInput, + DurableOrchestrationClient, + getGrpcHostAddress, +} from "./client"; +export { getClient, isDurableClientInput } from "./get-client"; +export { HttpManagementPayload } from "./http-management-payload"; +export { createAzureFunctionsMetadataGenerator } from "./metadata"; +export { DurableFunctionsWorker } from "./worker"; +export { DurableBindingMetadata, addDurableGrpcMetadata } from "./durable-grpc"; +export { RetryOptions } from "./retry-options"; +export { + DurableOrchestrationContext, + ClassicOrchestrationContext, + ClassicOrchestrator, + wrapOrchestrator, +} from "./orchestration-context"; +export { + DurableEntityContext, + ClassicEntityContext, + ClassicEntity, + wrapEntity, +} from "./entity-context"; +export { + DurableOrchestrationStatus, + DurableOrchestrationStatusInit, + OrchestrationRuntimeStatus, + fromOrchestrationRuntimeStatus, + toDurableOrchestrationStatus, + toOrchestrationRuntimeStatus, +} from "./orchestration-status"; +export { EntityId } from "./entity-id"; +export { EntityStateResponse } from "./entity-state-response"; +export { PurgeHistoryResult } from "./purge-history-result"; +export { OrchestrationFilter } from "./orchestration-filter"; + +// Legacy durable-functions v3 API compatibility aliases (types only). These let orchestrator/ +// activity code written against the classic `durable-functions` v3 API type-check unchanged. +export type { ActivityHandler } from "./app"; +export type { + ClassicOrchestrator as OrchestrationHandler, + ClassicOrchestrationContext as OrchestrationContext, +} from "./orchestration-context"; + +/** v3-compat: client returned by {@link getClient}. */ +export type DurableClient = DurableFunctionsClient; +/** + * v3-compat: entity context. `TState` is accepted for source compatibility with the legacy generic + * API; the underlying `df.getState()`/`getInput()` are per-call generic, so the type param is + * intentionally ignored here. + */ +export type EntityContext<_TState = unknown> = ClassicEntityContext; +/** v3-compat: entity handler. `TState` accepted for source compatibility (see {@link EntityContext}). */ +export type EntityHandler<_TState = unknown> = ClassicEntity; diff --git a/packages/azure-functions-durable/src/input.ts b/packages/azure-functions-durable/src/input.ts new file mode 100644 index 0000000..0b2e8f5 --- /dev/null +++ b/packages/azure-functions-durable/src/input.ts @@ -0,0 +1,14 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { FunctionInput, input as azFuncInput } from "@azure/functions"; +import { addDurableGrpcMetadata } from "./durable-grpc"; + +// The `durableClient` input binding hands the function the host-provided client configuration +// (task hub name, rpcBaseUrl, required query-string parameters, ...). Like the triggers, it opts +// in to the gRPC protocol via `durableRequiresGrpc: true` so the extension exposes its local +// sidecar; without it `rpcBaseUrl` is never populated and `getClient` cannot connect. + +export function durableClient(): FunctionInput { + return azFuncInput.generic(addDurableGrpcMetadata({ type: "durableClient" })); +} diff --git a/packages/azure-functions-durable/src/metadata.ts b/packages/azure-functions-durable/src/metadata.ts new file mode 100644 index 0000000..6f84a8b --- /dev/null +++ b/packages/azure-functions-durable/src/metadata.ts @@ -0,0 +1,19 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import * as grpc from "@grpc/grpc-js"; +import { MetadataGenerator } from "@microsoft/durabletask-js"; +import { getUserAgent } from "./user-agent"; + +export function createAzureFunctionsMetadataGenerator( + taskHubName: string, +): MetadataGenerator { + const userAgent = getUserAgent(); + + return async (): Promise => { + const metadata = new grpc.Metadata(); + metadata.set("taskhub", taskHubName); + metadata.set("x-user-agent", userAgent); + return metadata; + }; +} diff --git a/packages/azure-functions-durable/src/orchestration-context.ts b/packages/azure-functions-durable/src/orchestration-context.ts new file mode 100644 index 0000000..631f0c0 --- /dev/null +++ b/packages/azure-functions-durable/src/orchestration-context.ts @@ -0,0 +1,263 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { + ConsoleLogger, + EntityInstanceId, + OrchestrationContext, + Task, + TOrchestrator, + whenAll, + whenAny, +} from "@microsoft/durabletask-js"; +import { RetryOptions } from "./retry-options"; + +/** + * Classic Durable Functions (v3) orchestration context, exposed to migrating orchestrators as + * `context.df`. + * + * @remarks + * This is an adapter, not a re-implementation: every method forwards to the core durabletask + * {@link OrchestrationContext}. It lets orchestrator bodies written against the legacy + * `durable-functions` `context.df.*` API run unchanged on the gRPC/durabletask engine. + */ +export class DurableOrchestrationContext { + private _customStatus: unknown; + + constructor( + private readonly _ctx: OrchestrationContext, + private readonly _input: unknown, + ) {} + + /** The instance ID of the current orchestration. */ + get instanceId(): string { + return this._ctx.instanceId; + } + + /** The ID of the parent orchestration, or `undefined` if this is a top-level orchestration. */ + get parentInstanceId(): string | undefined { + return this._ctx.parent?.instanceId; + } + + /** Whether the orchestrator is currently replaying from history. */ + get isReplaying(): boolean { + return this._ctx.isReplaying; + } + + /** The replay-safe current UTC date/time. */ + get currentUtcDateTime(): Date { + return this._ctx.currentUtcDateTime; + } + + /** The version assigned to the current orchestration instance (empty string if none). */ + get version(): string { + return this._ctx.version; + } + + /** The custom status set during this execution via {@link setCustomStatus} (or `undefined`). */ + get customStatus(): unknown { + return this._customStatus; + } + + /** + * `Task.all` / `Task.any` fan-out/fan-in helpers (v3 shape), forwarding to the core + * `whenAll` / `whenAny` combinators. + */ + readonly Task = { + all: (tasks: Task[]): Task => whenAll(tasks), + any: (tasks: Task[]): Task> => whenAny(tasks) as unknown as Task>, + }; + + /** Gets the orchestration input. */ + getInput(): T { + return this._input as T; + } + + /** Schedules an activity for execution. */ + callActivity(name: string, input?: unknown): Task { + return this._ctx.callActivity(name, input); + } + + /** Schedules an activity for execution with a retry policy. */ + callActivityWithRetry( + name: string, + retryOptions: RetryOptions, + input?: unknown, + ): Task { + return this._ctx.callActivity(name, input, { + retry: retryOptions.toRetryPolicy(), + }); + } + + /** Schedules a sub-orchestrator for execution. */ + callSubOrchestrator( + name: string, + input?: unknown, + instanceId?: string, + version?: string, + ): Task { + return this._ctx.callSubOrchestrator( + name, + input, + instanceId !== undefined || version !== undefined ? { instanceId, version } : undefined, + ); + } + + /** Schedules a sub-orchestrator for execution with a retry policy. */ + callSubOrchestratorWithRetry( + name: string, + retryOptions: RetryOptions, + input?: unknown, + instanceId?: string, + version?: string, + ): Task { + return this._ctx.callSubOrchestrator(name, input, { + retry: retryOptions.toRetryPolicy(), + instanceId, + version, + }); + } + + /** Creates a durable timer that fires at the specified time. */ + createTimer(fireAt: Date | number): Task { + return this._ctx.createTimer(fireAt); + } + + /** Waits for an external event with the given name. */ + waitForExternalEvent(name: string): Task { + 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); + } + + /** Sets the orchestration's custom status payload. */ + setCustomStatus(customStatus: unknown): void { + this._customStatus = customStatus; + this._ctx.setCustomStatus(customStatus); + } + + /** + * Creates a new replay-safe UUID string. + * + * @param _instanceId - Accepted for classic v3 signature compatibility; ignored. The core engine + * derives the deterministic UUID from the instance ID and an internal sequence counter. + */ + newGuid(_instanceId?: string): string { + return this._ctx.newGuid(); + } + + /** + * Schedules a durable HTTP call (classic v3 API). + * + * @remarks + * Not supported: the durabletask engine has no durable-HTTP (`callHttp`) equivalent, so this + * throws. This mirrors the Python provider, which raises for the same reason. + */ + callHttp(_options: unknown): never { + throw new Error( + "callHttp is not supported: the durabletask engine has no durable-HTTP (callHttp) equivalent.", + ); + } + + /** Calls an entity operation and waits for its result. */ + callEntity( + entityId: EntityInstanceId, + operationName: string, + operationInput?: unknown, + ): Task { + return this._ctx.entities.callEntity(entityId, operationName, operationInput); + } + + /** Signals an entity operation (fire-and-forget). */ + signalEntity( + entityId: EntityInstanceId, + operationName: string, + operationInput?: unknown, + ): void { + this._ctx.entities.signalEntity(entityId, operationName, operationInput); + } +} + +/** The object passed to a classic (v3) orchestrator function; its `df` is the durable context. */ +export interface ClassicOrchestrationContext { + df: DurableOrchestrationContext; + /** Replay-safe log at info level (suppressed during replay). Alias of {@link info}. */ + log(...args: unknown[]): void; + /** Replay-safe error log (suppressed during replay). */ + error(...args: unknown[]): void; + /** Replay-safe warning log (suppressed during replay). */ + warn(...args: unknown[]): void; + /** Replay-safe info log (suppressed during replay). */ + info(...args: unknown[]): void; + /** Replay-safe debug log (suppressed during replay). */ + debug(...args: unknown[]): void; + /** Replay-safe trace log (mapped to debug; suppressed during replay). */ + trace(...args: unknown[]): void; +} + +/** + * A classic Durable Functions (v3) orchestrator: a single-argument generator function that reads + * and schedules work through `context.df.*`. + */ +export type ClassicOrchestrator = ( + context: ClassicOrchestrationContext, +) => Generator, unknown, unknown> | unknown; + +/** + * Adapts an orchestrator handler for registration on the core worker. + * + * @remarks + * Core-native orchestrators declare `(ctx, input)` and are returned unchanged. Classic v3 + * orchestrators declare a single `context` parameter and use `context.df.*`; those are wrapped so + * the core engine drives them while `context.df` forwards to the core {@link OrchestrationContext}. + * + * Detection is by arity: only single-parameter functions are treated as classic. Declare your + * core-native orchestrator as `(ctx, input)` so it passes through. + */ +export function wrapOrchestrator(handler: TOrchestrator | ClassicOrchestrator): TOrchestrator { + if (typeof handler === "function" && handler.length === 1) { + const classic = handler as ClassicOrchestrator; + // The core engine only drives orchestrators whose invocation returns an ASYNC generator + // (it gates on Symbol.asyncIterator). Classic v3 orchestrators are SYNC generators, so the + // wrapper must itself be an async generator that delegates to the classic one via `yield*`; + // that forwards each core Task to the engine and pipes sent results back into the classic body. + const wrapped = async function* ( + ctx: OrchestrationContext, + input: unknown, + ): AsyncGenerator, unknown, unknown> { + const df = new DurableOrchestrationContext(ctx, input); + // Replay-safe logger: output is suppressed while the engine replays history, matching the + // .NET/Python providers. Core Logger has no `trace`/plain `log`, so log→info and trace→debug. + const logger = ctx.createReplaySafeLogger(new ConsoleLogger()); + const toArgs = (a: unknown[]) => a as [string, ...unknown[]]; + const classicCtx: ClassicOrchestrationContext = { + df, + log: (...a) => logger.info(...toArgs(a)), + info: (...a) => logger.info(...toArgs(a)), + warn: (...a) => logger.warn(...toArgs(a)), + error: (...a) => logger.error(...toArgs(a)), + debug: (...a) => logger.debug(...toArgs(a)), + trace: (...a) => logger.debug(...toArgs(a)), + }; + const result = classic(classicCtx); + if (isGenerator(result)) { + return yield* result; + } + return result; + }; + return wrapped as unknown as TOrchestrator; + } + return handler as TOrchestrator; +} + +/** @hidden */ +function isGenerator(value: unknown): value is Generator, unknown, unknown> { + return ( + value != null && + typeof (value as Iterator).next === "function" && + typeof (value as { [Symbol.iterator]?: unknown })[Symbol.iterator] === "function" + ); +} diff --git a/packages/azure-functions-durable/src/orchestration-filter.ts b/packages/azure-functions-durable/src/orchestration-filter.ts new file mode 100644 index 0000000..e78bda6 --- /dev/null +++ b/packages/azure-functions-durable/src/orchestration-filter.ts @@ -0,0 +1,18 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { OrchestrationRuntimeStatus } from "./orchestration-status"; + +/** + * Filter for selecting orchestration instances by creation time and/or runtime status, + * matching the classic Durable Functions v3 `OrchestrationFilter` shape. Used by + * {@link DurableFunctionsClient.getStatusBy} and {@link DurableFunctionsClient.purgeInstanceHistoryBy}. + */ +export interface OrchestrationFilter { + /** Select orchestration instances created at or after this time. */ + createdTimeFrom?: Date; + /** Select orchestration instances created at or before this time. */ + createdTimeTo?: Date; + /** Select orchestration instances whose runtime status matches any value in this array. */ + runtimeStatus?: OrchestrationRuntimeStatus[]; +} diff --git a/packages/azure-functions-durable/src/orchestration-status.ts b/packages/azure-functions-durable/src/orchestration-status.ts new file mode 100644 index 0000000..afd7678 --- /dev/null +++ b/packages/azure-functions-durable/src/orchestration-status.ts @@ -0,0 +1,127 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { OrchestrationState, OrchestrationStatus } from "@microsoft/durabletask-js"; + +/** + * The runtime status of an orchestration instance (classic Durable Functions v3 shape). + */ +export enum OrchestrationRuntimeStatus { + Running = "Running", + Completed = "Completed", + ContinuedAsNew = "ContinuedAsNew", + Failed = "Failed", + Canceled = "Canceled", + Terminated = "Terminated", + Pending = "Pending", + Suspended = "Suspended", +} + +/** Fields used to construct a {@link DurableOrchestrationStatus}. */ +export interface DurableOrchestrationStatusInit { + name: string; + instanceId: string; + createdTime: Date; + lastUpdatedTime: Date; + input: unknown; + output: unknown; + runtimeStatus: OrchestrationRuntimeStatus; + customStatus?: unknown; + history?: unknown[]; +} + +/** + * Represents the status of a durable orchestration instance, matching the classic Durable Functions + * v3 `DurableOrchestrationStatus` shape. Returned by {@link DurableFunctionsClient.getStatus}. + */ +export class DurableOrchestrationStatus { + readonly name: string; + readonly instanceId: string; + readonly createdTime: Date; + readonly lastUpdatedTime: Date; + readonly input: unknown; + readonly output: unknown; + readonly runtimeStatus: OrchestrationRuntimeStatus; + readonly customStatus?: unknown; + readonly history?: unknown[]; + + constructor(init: DurableOrchestrationStatusInit) { + this.name = init.name; + this.instanceId = init.instanceId; + this.createdTime = init.createdTime; + this.lastUpdatedTime = init.lastUpdatedTime; + this.input = init.input; + this.output = init.output; + this.runtimeStatus = init.runtimeStatus; + this.customStatus = init.customStatus; + this.history = init.history; + } +} + +/** Maps the core {@link OrchestrationStatus} enum to the classic v3 {@link OrchestrationRuntimeStatus}. */ +export function toOrchestrationRuntimeStatus(status: OrchestrationStatus): OrchestrationRuntimeStatus { + switch (status) { + case OrchestrationStatus.RUNNING: + return OrchestrationRuntimeStatus.Running; + case OrchestrationStatus.COMPLETED: + return OrchestrationRuntimeStatus.Completed; + case OrchestrationStatus.CONTINUED_AS_NEW: + return OrchestrationRuntimeStatus.ContinuedAsNew; + case OrchestrationStatus.FAILED: + return OrchestrationRuntimeStatus.Failed; + case OrchestrationStatus.TERMINATED: + return OrchestrationRuntimeStatus.Terminated; + case OrchestrationStatus.PENDING: + return OrchestrationRuntimeStatus.Pending; + case OrchestrationStatus.SUSPENDED: + return OrchestrationRuntimeStatus.Suspended; + default: + return OrchestrationRuntimeStatus.Pending; + } +} + +/** Maps a classic v3 {@link OrchestrationRuntimeStatus} back to the core {@link OrchestrationStatus}. */ +export function fromOrchestrationRuntimeStatus(status: OrchestrationRuntimeStatus): OrchestrationStatus { + switch (status) { + case OrchestrationRuntimeStatus.Running: + return OrchestrationStatus.RUNNING; + case OrchestrationRuntimeStatus.Completed: + return OrchestrationStatus.COMPLETED; + case OrchestrationRuntimeStatus.ContinuedAsNew: + return OrchestrationStatus.CONTINUED_AS_NEW; + case OrchestrationRuntimeStatus.Failed: + return OrchestrationStatus.FAILED; + case OrchestrationRuntimeStatus.Terminated: + case OrchestrationRuntimeStatus.Canceled: + return OrchestrationStatus.TERMINATED; + case OrchestrationRuntimeStatus.Suspended: + return OrchestrationStatus.SUSPENDED; + case OrchestrationRuntimeStatus.Pending: + default: + return OrchestrationStatus.PENDING; + } +} + +/** + * Maps a core {@link OrchestrationState} into the classic v3 {@link DurableOrchestrationStatus}. + * + * Payloads are deserialized with `JSON.parse`, matching the plain-JSON wire contract the core client + * and worker use. + */ +export function toDurableOrchestrationStatus(state: OrchestrationState): DurableOrchestrationStatus { + return new DurableOrchestrationStatus({ + name: state.name, + instanceId: state.instanceId, + createdTime: state.createdAt, + lastUpdatedTime: state.lastUpdatedAt, + input: parseJson(state.serializedInput), + output: parseJson(state.serializedOutput), + runtimeStatus: toOrchestrationRuntimeStatus(state.runtimeStatus), + customStatus: parseJson(state.serializedCustomStatus), + }); +} + +/** @hidden */ +function parseJson(value: string | undefined): unknown { + return value === undefined ? undefined : JSON.parse(value); +} diff --git a/packages/azure-functions-durable/src/purge-history-result.ts b/packages/azure-functions-durable/src/purge-history-result.ts new file mode 100644 index 0000000..987c68b --- /dev/null +++ b/packages/azure-functions-durable/src/purge-history-result.ts @@ -0,0 +1,13 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +/** + * Statistics about a purge-history operation, matching the classic Durable Functions v3 + * `PurgeHistoryResult` shape. Returned by {@link DurableFunctionsClient.purgeInstanceHistory}. + */ +export class PurgeHistoryResult { + /** + * @param instancesDeleted - The number of orchestration instances that were purged. + */ + constructor(public readonly instancesDeleted: number) {} +} diff --git a/packages/azure-functions-durable/src/retry-options.ts b/packages/azure-functions-durable/src/retry-options.ts new file mode 100644 index 0000000..d86b1d2 --- /dev/null +++ b/packages/azure-functions-durable/src/retry-options.ts @@ -0,0 +1,46 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { RetryPolicy } from "@microsoft/durabletask-js"; + +/** + * Classic Durable Functions (v3) retry configuration for `callActivityWithRetry` / + * `callSubOrchestratorWithRetry`. + * + * @remarks + * This mirrors the legacy `durable-functions` `RetryOptions` shape so existing orchestrator code + * migrates without changes. It is a thin adapter: {@link toRetryPolicy} converts it to the core + * `RetryPolicy` that the durabletask engine actually understands. + */ +export class RetryOptions { + /** Exponential backoff coefficient applied between retries. Defaults to 1 (no backoff). */ + public backoffCoefficient = 1; + + /** Maximum delay between retries, in milliseconds. When unset, the core default (1 hour) applies. */ + public maxRetryIntervalInMilliseconds?: number; + + /** Overall retry timeout, in milliseconds. When unset, retries are not time-bounded. */ + public retryTimeoutInMilliseconds?: number; + + /** + * @param firstRetryIntervalInMilliseconds - Delay before the first retry, in milliseconds. + * @param maxNumberOfAttempts - Maximum number of attempts (including the first call). + */ + constructor( + public readonly firstRetryIntervalInMilliseconds: number, + public readonly maxNumberOfAttempts: number, + ) {} + + /** + * Converts this classic `RetryOptions` into the core {@link RetryPolicy}. + */ + toRetryPolicy(): RetryPolicy { + return new RetryPolicy({ + firstRetryIntervalInMilliseconds: this.firstRetryIntervalInMilliseconds, + maxNumberOfAttempts: this.maxNumberOfAttempts, + backoffCoefficient: this.backoffCoefficient, + maxRetryIntervalInMilliseconds: this.maxRetryIntervalInMilliseconds, + retryTimeoutInMilliseconds: this.retryTimeoutInMilliseconds, + }); + } +} diff --git a/packages/azure-functions-durable/src/trigger.ts b/packages/azure-functions-durable/src/trigger.ts new file mode 100644 index 0000000..e518a96 --- /dev/null +++ b/packages/azure-functions-durable/src/trigger.ts @@ -0,0 +1,22 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { FunctionTrigger, trigger as azFuncTrigger } from "@azure/functions"; +import { addDurableGrpcMetadata } from "./durable-grpc"; + +// The `type` strings below are the wire contract expected by the Durable Task extension's +// binding providers; they must not be renamed. Every Durable trigger opts in to the extension's +// local gRPC protocol by carrying `durableRequiresGrpc: true` (see `addDurableGrpcMetadata`), +// which is what makes the extension open its local sidecar and populate `rpcBaseUrl` for the client. + +export function orchestration(): FunctionTrigger { + return azFuncTrigger.generic(addDurableGrpcMetadata({ type: "orchestrationTrigger" })); +} + +export function activity(): FunctionTrigger { + return azFuncTrigger.generic(addDurableGrpcMetadata({ type: "activityTrigger" })); +} + +export function entity(): FunctionTrigger { + return azFuncTrigger.generic(addDurableGrpcMetadata({ type: "entityTrigger" })); +} diff --git a/packages/azure-functions-durable/src/user-agent.ts b/packages/azure-functions-durable/src/user-agent.ts new file mode 100644 index 0000000..fe69113 --- /dev/null +++ b/packages/azure-functions-durable/src/user-agent.ts @@ -0,0 +1,23 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +const SDK_NAME = "durable-functions"; + +let packageVersion = "unknown"; + +function getPackageVersion(): string { + if (packageVersion === "unknown") { + try { + const pkg = require("../package.json"); + packageVersion = pkg.version ?? "unknown"; + } catch { + // Keep the fallback when package metadata is unavailable. + } + } + + return packageVersion; +} + +export function getUserAgent(): string { + return `${SDK_NAME}/${getPackageVersion()}`; +} diff --git a/packages/azure-functions-durable/src/worker.ts b/packages/azure-functions-durable/src/worker.ts new file mode 100644 index 0000000..4eb2bc8 --- /dev/null +++ b/packages/azure-functions-durable/src/worker.ts @@ -0,0 +1,30 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { TaskHubGrpcWorker, TaskHubGrpcWorkerOptions } from "@microsoft/durabletask-js"; + +export class DurableFunctionsWorker extends TaskHubGrpcWorker { + constructor(options: TaskHubGrpcWorkerOptions = {}) { + super(options); + } + + async handleOrchestratorRequest(encodedRequest: string): Promise { + const request = decodeBase64Request(encodedRequest, "orchestrator"); + const response = await this.processOrchestratorRequest(request); + return Buffer.from(response).toString("base64"); + } + + async handleEntityBatchRequest(encodedRequest: string): Promise { + const request = decodeBase64Request(encodedRequest, "entity batch"); + const response = await this.processEntityBatchRequest(request); + return Buffer.from(response).toString("base64"); + } +} + +function decodeBase64Request(encodedRequest: string, requestType: string): Buffer { + if (!encodedRequest) { + throw new TypeError(`${requestType} request must be a non-empty base64 string.`); + } + + return Buffer.from(encodedRequest, "base64"); +} diff --git a/packages/azure-functions-durable/test/unit/app.spec.ts b/packages/azure-functions-durable/test/unit/app.spec.ts new file mode 100644 index 0000000..8c5369a --- /dev/null +++ b/packages/azure-functions-durable/test/unit/app.spec.ts @@ -0,0 +1,85 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { GenericFunctionOptions, InvocationContext, app as azFuncApp } from "@azure/functions"; +import * as app from "../../src/app"; +import { DurableFunctionsWorker } from "../../src/worker"; + +describe("app registration", () => { + let genericSpy: jest.SpyInstance; + + beforeEach(() => { + genericSpy = jest.spyOn(azFuncApp, "generic").mockImplementation(() => undefined); + }); + + afterEach(() => { + jest.restoreAllMocks(); + }); + + function lastRegistration(): { name: string; options: GenericFunctionOptions } { + const call = genericSpy.mock.calls[genericSpy.mock.calls.length - 1]; + return { name: call[0] as string, options: call[1] as GenericFunctionOptions }; + } + + it("registers an orchestration whose trigger opts in to gRPC and whose handler forwards base64", async () => { + const handleSpy = jest + .spyOn(DurableFunctionsWorker.prototype, "handleOrchestratorRequest") + .mockResolvedValue("encoded-orchestrator-response"); + + app.orchestration("orchestration-test", () => undefined); + + const { name, options } = lastRegistration(); + expect(name).toBe("orchestration-test"); + expect(options.trigger.type).toBe("orchestrationTrigger"); + expect(options.trigger.durableRequiresGrpc).toBe(true); + + const result = await options.handler("base64-orchestrator-request", {} as InvocationContext); + + expect(result).toBe("encoded-orchestrator-response"); + expect(handleSpy).toHaveBeenCalledWith("base64-orchestrator-request"); + }); + + it("registers an entity whose trigger opts in to gRPC and whose handler forwards base64", async () => { + const handleSpy = jest + .spyOn(DurableFunctionsWorker.prototype, "handleEntityBatchRequest") + .mockResolvedValue("encoded-entity-response"); + + app.entity("entity-test", () => ({}) as never); + + const { name, options } = lastRegistration(); + expect(name).toBe("entity-test"); + expect(options.trigger.type).toBe("entityTrigger"); + expect(options.trigger.durableRequiresGrpc).toBe(true); + + const result = await options.handler("base64-entity-request", {} as InvocationContext); + + expect(result).toBe("encoded-entity-response"); + expect(handleSpy).toHaveBeenCalledWith("base64-entity-request"); + }); + + it("registers an activity as a plain pass-through handler that still opts in to gRPC", async () => { + const userHandler = jest.fn().mockResolvedValue("activity-result"); + + app.activity("activity-test", userHandler); + + const { name, options } = lastRegistration(); + expect(name).toBe("activity-test"); + expect(options.trigger.type).toBe("activityTrigger"); + expect(options.trigger.durableRequiresGrpc).toBe(true); + expect(options.handler).toBe(userHandler); + }); + + it("throws when a durable trigger delivers no base64 request body", async () => { + jest + .spyOn(DurableFunctionsWorker.prototype, "handleOrchestratorRequest") + .mockResolvedValue("unused"); + + app.orchestration("orchestration-missing-body", { + handler: () => undefined, + }); + + const { options } = lastRegistration(); + + await expect(options.handler(undefined, {} as InvocationContext)).rejects.toThrow(TypeError); + }); +}); diff --git a/packages/azure-functions-durable/test/unit/client-query.spec.ts b/packages/azure-functions-durable/test/unit/client-query.spec.ts new file mode 100644 index 0000000..b51a977 --- /dev/null +++ b/packages/azure-functions-durable/test/unit/client-query.spec.ts @@ -0,0 +1,208 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { EntityInstanceId, OrchestrationState, OrchestrationStatus } from "@microsoft/durabletask-js"; +import { HttpRequest } from "@azure/functions"; +import { DurableFunctionsClient } from "../../src/client"; +import { DurableOrchestrationStatus, OrchestrationRuntimeStatus } from "../../src/orchestration-status"; +import { EntityStateResponse } from "../../src/entity-state-response"; +import { PurgeHistoryResult } from "../../src/purge-history-result"; + +const CLIENT_CONFIG = { + taskHubName: "functions-taskhub", + rpcBaseUrl: "http://127.0.0.1:4711/rpc", + requiredQueryStringParameters: "code=secret", +}; + +function stateOf(instanceId: string, status = OrchestrationStatus.RUNNING): OrchestrationState { + return new OrchestrationState( + instanceId, + "orch", + status, + new Date("2026-01-01T00:00:00.000Z"), + new Date("2026-01-01T00:00:00.000Z"), + ); +} + +async function* asyncGen(items: T[]): AsyncGenerator { + for (const item of items) { + yield item; + } +} + +describe("DurableFunctionsClient query methods", () => { + it("getStatus maps the core state to a DurableOrchestrationStatus", async () => { + const client = new DurableFunctionsClient(CLIENT_CONFIG); + try { + const state = new OrchestrationState( + "abc", + "helloOrchestrator", + OrchestrationStatus.RUNNING, + new Date("2026-01-01T00:00:00.000Z"), + new Date("2026-01-01T00:00:00.000Z"), + JSON.stringify({ n: 1 }), + ); + const spy = jest.spyOn(client, "getOrchestrationState").mockResolvedValue(state); + + const status = await client.getStatus("abc"); + + expect(spy).toHaveBeenCalledWith("abc", true); + expect(status).toBeInstanceOf(DurableOrchestrationStatus); + expect(status?.instanceId).toBe("abc"); + expect(status?.input).toEqual({ n: 1 }); + expect(status?.runtimeStatus).toBe(OrchestrationRuntimeStatus.Running); + } finally { + await client.stop(); + } + }); + + it("getStatus returns undefined when the instance does not exist", async () => { + const client = new DurableFunctionsClient(CLIENT_CONFIG); + try { + jest.spyOn(client, "getOrchestrationState").mockResolvedValue(undefined); + await expect(client.getStatus("missing")).resolves.toBeUndefined(); + } finally { + await client.stop(); + } + }); + + it("readEntityState maps present and absent entities", async () => { + const client = new DurableFunctionsClient(CLIENT_CONFIG); + const entityId = new EntityInstanceId("Counter", "k1"); + try { + const getEntity = jest.spyOn(client, "getEntity"); + + getEntity.mockResolvedValueOnce({ + id: entityId, + lastModifiedTime: new Date(), + backlogQueueSize: 0, + includesState: true, + state: 42, + }); + const present = await client.readEntityState(entityId); + expect(present).toBeInstanceOf(EntityStateResponse); + expect(present.entityExists).toBe(true); + expect(present.entityState).toBe(42); + + getEntity.mockResolvedValueOnce(undefined); + const absent = await client.readEntityState(entityId); + expect(absent.entityExists).toBe(false); + expect(absent.entityState).toBeUndefined(); + } finally { + await client.stop(); + } + }); + + it("purgeInstanceHistory maps the core purge result", async () => { + const client = new DurableFunctionsClient(CLIENT_CONFIG); + try { + jest + .spyOn(client, "purgeOrchestration") + .mockResolvedValue({ deletedInstanceCount: 1 } as never); + + const result = await client.purgeInstanceHistory("abc"); + expect(result).toBeInstanceOf(PurgeHistoryResult); + expect(result.instancesDeleted).toBe(1); + } finally { + await client.stop(); + } + }); + + it("startNew forwards to scheduleNewOrchestration with input and instanceId", async () => { + const client = new DurableFunctionsClient(CLIENT_CONFIG); + try { + const schedule = jest.spyOn(client, "scheduleNewOrchestration").mockResolvedValue("new-id"); + + await client.startNew("hello", { input: { city: "Tokyo" }, instanceId: "id-1" }); + expect(schedule).toHaveBeenCalledWith("hello", { city: "Tokyo" }, { instanceId: "id-1" }); + + await client.startNew("hello"); + expect(schedule).toHaveBeenLastCalledWith("hello", undefined, undefined); + } finally { + await client.stop(); + } + }); + + it("getStatusAll maps every instance from the paged query", async () => { + const client = new DurableFunctionsClient(CLIENT_CONFIG); + try { + const getAll = jest + .spyOn(client, "getAllInstances") + .mockReturnValue(asyncGen([stateOf("a"), stateOf("b")]) as never); + + const all = await client.getStatusAll(); + expect(getAll).toHaveBeenCalledWith({ fetchInputsAndOutputs: true }); + expect(all.map((s) => s.instanceId)).toEqual(["a", "b"]); + expect(all[0]).toBeInstanceOf(DurableOrchestrationStatus); + } finally { + await client.stop(); + } + }); + + it("getStatusBy maps the v3 filter to a core query", async () => { + const client = new DurableFunctionsClient(CLIENT_CONFIG); + try { + const getAll = jest + .spyOn(client, "getAllInstances") + .mockReturnValue(asyncGen([stateOf("a", OrchestrationStatus.COMPLETED)]) as never); + + const from = new Date("2026-01-01T00:00:00.000Z"); + const to = new Date("2026-01-02T00:00:00.000Z"); + const result = await client.getStatusBy({ + createdTimeFrom: from, + createdTimeTo: to, + runtimeStatus: [OrchestrationRuntimeStatus.Completed], + }); + + expect(getAll).toHaveBeenCalledWith({ + createdFrom: from, + createdTo: to, + statuses: [OrchestrationStatus.COMPLETED], + fetchInputsAndOutputs: true, + }); + expect(result[0].runtimeStatus).toBe(OrchestrationRuntimeStatus.Completed); + } finally { + await client.stop(); + } + }); + + it("waitForCompletionOrCreateCheckStatusResponse returns 200 with output when completed", async () => { + const client = new DurableFunctionsClient(CLIENT_CONFIG); + const request = new HttpRequest({ method: "GET", url: "http://localhost:7071/api/x" }); + try { + const completed = new OrchestrationState( + "abc", + "orch", + OrchestrationStatus.COMPLETED, + new Date(), + new Date(), + undefined, + JSON.stringify("the-output"), + ); + jest.spyOn(client, "waitForOrchestrationCompletion").mockResolvedValue(completed); + + const response = await client.waitForCompletionOrCreateCheckStatusResponse(request, "abc"); + expect(response.status).toBe(200); + expect(await response.text()).toBe(JSON.stringify("the-output")); + } finally { + await client.stop(); + } + }); + + it("waitForCompletionOrCreateCheckStatusResponse falls back to 202 on timeout", async () => { + const client = new DurableFunctionsClient(CLIENT_CONFIG); + const request = new HttpRequest({ method: "GET", url: "http://localhost:7071/api/x" }); + try { + jest + .spyOn(client, "waitForOrchestrationCompletion") + .mockRejectedValue(new Error("Timed out")); + + const response = await client.waitForCompletionOrCreateCheckStatusResponse(request, "abc", { + timeoutInMilliseconds: 1000, + }); + expect(response.status).toBe(202); + } finally { + await client.stop(); + } + }); +}); diff --git a/packages/azure-functions-durable/test/unit/client.spec.ts b/packages/azure-functions-durable/test/unit/client.spec.ts new file mode 100644 index 0000000..f68ff83 --- /dev/null +++ b/packages/azure-functions-durable/test/unit/client.spec.ts @@ -0,0 +1,235 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { HttpRequest } from "@azure/functions"; +import { PurgeInstanceCriteria, TaskHubGrpcClient } from "@microsoft/durabletask-js"; +import { DurableFunctionsClient, getGrpcHostAddress } from "../../src/client"; +import { createAzureFunctionsMetadataGenerator } from "../../src/metadata"; +import { OrchestrationRuntimeStatus } from "../../src/orchestration-status"; + +const CLIENT_CONFIG = { + taskHubName: "functions-taskhub", + rpcBaseUrl: "http://127.0.0.1:4711/rpc", + requiredQueryStringParameters: "code=secret&taskHub=functions-taskhub", + httpBaseUrl: "https://ignored.example/runtime/webhooks/durabletask", +}; + +describe("DurableFunctionsClient", () => { + it("derives the gRPC host address from rpcBaseUrl", () => { + expect(getGrpcHostAddress("http://127.0.0.1:4711/rpc")).toBe("127.0.0.1:4711"); + expect(getGrpcHostAddress("https://localhost:9443")).toBe("localhost:9443"); + }); + + it("extends TaskHubGrpcClient and does not redeclare management methods", async () => { + const client = new DurableFunctionsClient(JSON.stringify(CLIENT_CONFIG)); + + try { + expect(client).toBeInstanceOf(TaskHubGrpcClient); + expect(typeof client.scheduleNewOrchestration).toBe("function"); + expect(typeof client.getOrchestrationState).toBe("function"); + expect(typeof client.raiseOrchestrationEvent).toBe("function"); + expect(typeof client.terminateOrchestration).toBe("function"); + expect(typeof client.suspendOrchestration).toBe("function"); + expect(typeof client.resumeOrchestration).toBe("function"); + expect(typeof client.purgeOrchestration).toBe("function"); + expect(typeof client.signalEntity).toBe("function"); + expect(typeof client.getEntity).toBe("function"); + expect(typeof client.raiseEvent).toBe("function"); + expect(typeof client.terminate).toBe("function"); + expect(typeof client.suspend).toBe("function"); + expect(typeof client.resume).toBe("function"); + expect(typeof client.rewind).toBe("function"); + expect(typeof client.restart).toBe("function"); + expect(typeof client.purgeInstanceHistoryBy).toBe("function"); + expect(Object.getOwnPropertyNames(DurableFunctionsClient.prototype).sort()).toEqual([ + "collectStatuses", + "constructor", + "createCheckStatusResponse", + "createHttpManagementPayload", + "getClientResponseLinks", + "getStatus", + "getStatusAll", + "getStatusBy", + "purgeInstanceHistory", + "purgeInstanceHistoryBy", + "raiseEvent", + "readEntityState", + "restart", + "resume", + "rewind", + "startNew", + "suspend", + "terminate", + "waitForCompletionOrCreateCheckStatusResponse", + ]); + } finally { + await client.stop(); + } + }); + + it("forwards classic Durable Functions v3 lifecycle aliases to the core methods", async () => { + const client = new DurableFunctionsClient(CLIENT_CONFIG); + + try { + const raise = jest.spyOn(client, "raiseOrchestrationEvent").mockResolvedValue(undefined); + await client.raiseEvent("id-1", "evt", { a: 1 }); + expect(raise).toHaveBeenCalledWith("id-1", "evt", { a: 1 }); + + const terminate = jest.spyOn(client, "terminateOrchestration").mockResolvedValue(undefined); + await client.terminate("id-1", "cancelled"); + expect(terminate).toHaveBeenCalledWith("id-1", "cancelled"); + + const suspend = jest.spyOn(client, "suspendOrchestration").mockResolvedValue(undefined); + await client.suspend("id-1", "ignored-reason"); + expect(suspend).toHaveBeenCalledWith("id-1"); + + const resume = jest.spyOn(client, "resumeOrchestration").mockResolvedValue(undefined); + await client.resume("id-1", "ignored-reason"); + expect(resume).toHaveBeenCalledWith("id-1"); + + await expect(client.rewind("id-1", "retrying")).rejects.toThrow( + "rewind is not yet supported", + ); + + const restart = jest.spyOn(client, "restartOrchestration").mockResolvedValue("id-2"); + await expect(client.restart("id-1", true)).resolves.toBe("id-2"); + expect(restart).toHaveBeenCalledWith("id-1", true); + + const purge = jest.spyOn(client, "purgeOrchestration").mockResolvedValue(undefined); + const purgeResult = await client.purgeInstanceHistoryBy({ + createdTimeFrom: new Date("2020-01-01T00:00:00.000Z"), + createdTimeTo: new Date("2020-02-01T00:00:00.000Z"), + runtimeStatus: [OrchestrationRuntimeStatus.Completed], + }); + expect(purgeResult.instancesDeleted).toBe(0); + expect(purge).toHaveBeenCalledTimes(1); + const criteria = purge.mock.calls[0][0] as PurgeInstanceCriteria; + expect(criteria.getCreatedTimeFrom()).toEqual(new Date("2020-01-01T00:00:00.000Z")); + expect(criteria.getCreatedTimeTo()).toEqual(new Date("2020-02-01T00:00:00.000Z")); + expect(criteria.getRuntimeStatusList()).toHaveLength(1); + } finally { + await client.stop(); + } + }); + + it("creates HTTP management payload URLs from the incoming request", async () => { + const client = new DurableFunctionsClient(CLIENT_CONFIG); + const request = new HttpRequest({ + method: "POST", + url: "https://public.example/api/start?ignored=true", + }); + + try { + const payload = client.createHttpManagementPayload(request, "instance 1"); + + expect(payload).toEqual({ + id: "instance 1", + purgeHistoryDeleteUri: + "https://public.example/runtime/webhooks/durabletask/instances/instance%201?code=secret&taskHub=functions-taskhub", + restartPostUri: + "https://public.example/runtime/webhooks/durabletask/instances/instance%201/restart?code=secret&taskHub=functions-taskhub", + rewindPostUri: + "https://public.example/runtime/webhooks/durabletask/instances/instance%201/rewind?reason={text}&code=secret&taskHub=functions-taskhub", + sendEventPostUri: + "https://public.example/runtime/webhooks/durabletask/instances/instance%201/raiseEvent/{eventName}?code=secret&taskHub=functions-taskhub", + statusQueryGetUri: + "https://public.example/runtime/webhooks/durabletask/instances/instance%201?code=secret&taskHub=functions-taskhub", + terminatePostUri: + "https://public.example/runtime/webhooks/durabletask/instances/instance%201/terminate?reason={text}&code=secret&taskHub=functions-taskhub", + resumePostUri: + "https://public.example/runtime/webhooks/durabletask/instances/instance%201/resume?reason={text}&code=secret&taskHub=functions-taskhub", + suspendPostUri: + "https://public.example/runtime/webhooks/durabletask/instances/instance%201/suspend?reason={text}&code=secret&taskHub=functions-taskhub", + }); + } finally { + await client.stop(); + } + }); + + it("creates HTTP management payload URLs from baseUrl for the classic v3 single-argument call", async () => { + const client = new DurableFunctionsClient({ + ...CLIENT_CONFIG, + baseUrl: "https://public.example/runtime/webhooks/durabletask/", + }); + + try { + const payload = client.createHttpManagementPayload("instance 1"); + + expect(payload.id).toBe("instance 1"); + expect(payload.statusQueryGetUri).toBe( + "https://public.example/runtime/webhooks/durabletask/instances/instance%201?code=secret&taskHub=functions-taskhub", + ); + expect(payload.terminatePostUri).toBe( + "https://public.example/runtime/webhooks/durabletask/instances/instance%201/terminate?reason={text}&code=secret&taskHub=functions-taskhub", + ); + } finally { + await client.stop(); + } + }); + + it("throws when createHttpManagementPayload is called without an instance id", async () => { + const client = new DurableFunctionsClient(CLIENT_CONFIG); + const request = new HttpRequest({ + method: "POST", + url: "https://public.example/api/start", + }); + + try { + expect(() => + (client.createHttpManagementPayload as unknown as (request: HttpRequest) => unknown)( + request, + ), + ).toThrow(TypeError); + } finally { + await client.stop(); + } + }); + + it("exposes getClientResponseLinks as a deprecated alias of createHttpManagementPayload", async () => { + const client = new DurableFunctionsClient(CLIENT_CONFIG); + const request = new HttpRequest({ + method: "POST", + url: "https://public.example/api/start", + }); + + try { + expect(client.getClientResponseLinks(request, "abc")).toEqual( + client.createHttpManagementPayload(request, "abc"), + ); + } finally { + await client.stop(); + } + }); + + it("creates 202 check status responses with Location and JSON body", async () => { + const client = new DurableFunctionsClient(CLIENT_CONFIG); + const request = new HttpRequest({ + method: "POST", + url: "http://localhost:7071/api/orchestrators/hello", + }); + + try { + const response = client.createCheckStatusResponse(request, "abc"); + + expect(response.status).toBe(202); + expect(response.headers.get("content-type")).toBe("application/json"); + expect(response.headers.get("Location")).toBe( + "http://localhost:7071/runtime/webhooks/durabletask/instances/abc?code=secret&taskHub=functions-taskhub", + ); + const body = JSON.parse(await response.text()); + expect(body.statusQueryGetUri).toBe( + "http://localhost:7071/runtime/webhooks/durabletask/instances/abc?code=secret&taskHub=functions-taskhub", + ); + } finally { + await client.stop(); + } + }); + + it("mirrors the Azure Functions gRPC metadata interceptor", async () => { + const metadata = await createAzureFunctionsMetadataGenerator("functions-taskhub")(); + + expect(metadata.get("taskhub")).toEqual(["functions-taskhub"]); + expect(metadata.get("x-user-agent")[0]).toMatch(/^durable-functions\//); + expect(metadata.get("requiredQueryStringParameters")).toEqual([]); + }); +}); diff --git a/packages/azure-functions-durable/test/unit/compat-exports.spec.ts b/packages/azure-functions-durable/test/unit/compat-exports.spec.ts new file mode 100644 index 0000000..918e962 --- /dev/null +++ b/packages/azure-functions-durable/test/unit/compat-exports.spec.ts @@ -0,0 +1,37 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import type { + ActivityHandler, + DurableClient, + EntityContext, + EntityHandler, + OrchestrationContext, + OrchestrationHandler, +} from "../../src"; + +describe("v3 compatibility type aliases", () => { + it("exposes ActivityHandler / OrchestrationHandler / OrchestrationContext", () => { + const activity: ActivityHandler = (input: unknown) => input; + const orchestrator: OrchestrationHandler = function* (context: OrchestrationContext) { + yield context.df.callActivity("noop"); + return context.df.getInput(); + }; + expect(typeof activity).toBe("function"); + expect(typeof orchestrator).toBe("function"); + }); + + it("exposes generic EntityContext / EntityHandler and DurableClient", () => { + // Compile-guard: the generic aliases must accept a type argument (the legacy v3 surface uses + // e.g. EntityHandler), even though our underlying types are non-generic. + type _E1 = EntityContext<{ x: number }>; + type _E2 = EntityHandler; + const handler: EntityHandler = (context: EntityContext) => { + context.df.return(0); + }; + // DurableClient is the type returned by getClient(); assert it's usable as a type annotation. + const client: DurableClient | undefined = undefined; + expect(typeof handler).toBe("function"); + expect(client).toBeUndefined(); + }); +}); diff --git a/packages/azure-functions-durable/test/unit/durable-grpc.spec.ts b/packages/azure-functions-durable/test/unit/durable-grpc.spec.ts new file mode 100644 index 0000000..89d01ac --- /dev/null +++ b/packages/azure-functions-durable/test/unit/durable-grpc.spec.ts @@ -0,0 +1,19 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { addDurableGrpcMetadata } from "../../src/durable-grpc"; + +describe("addDurableGrpcMetadata", () => { + it("adds durableRequiresGrpc without mutating the original binding", () => { + const binding = { type: "orchestrationTrigger", name: "context" }; + + const actual = addDurableGrpcMetadata(binding); + + expect(actual).toEqual({ + type: "orchestrationTrigger", + name: "context", + durableRequiresGrpc: true, + }); + expect(binding).toEqual({ type: "orchestrationTrigger", name: "context" }); + }); +}); diff --git a/packages/azure-functions-durable/test/unit/entity-context.spec.ts b/packages/azure-functions-durable/test/unit/entity-context.spec.ts new file mode 100644 index 0000000..2044ad7 --- /dev/null +++ b/packages/azure-functions-durable/test/unit/entity-context.spec.ts @@ -0,0 +1,154 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { EntityFactory, EntityInstanceId, TaskEntityOperation } from "@microsoft/durabletask-js"; +import { + ClassicEntityContext, + DurableEntityContext, + wrapEntity, +} from "../../src/entity-context"; + +/** Builds a fake core TaskEntityOperation backed by an in-memory state cell. */ +function createFakeOperation(options: { + name: string; + input?: unknown; + initialState?: unknown; +}): { operation: TaskEntityOperation; getStoredState: () => unknown } { + let stored: unknown = options.initialState; + let hasState = options.initialState !== undefined; + + const operation = { + name: options.name, + context: { + id: new EntityInstanceId("Counter", "user-1"), + signalEntity: jest.fn(), + scheduleNewOrchestration: jest.fn(), + }, + state: { + get hasState() { + return hasState; + }, + getState(defaultValue?: T): T | undefined { + return (hasState ? (stored as T) : defaultValue) as T | undefined; + }, + setState(state: unknown): void { + stored = state; + hasState = state !== undefined && state !== null; + }, + }, + hasInput: options.input !== undefined, + getInput(): T | undefined { + return options.input as T | undefined; + }, + }; + + return { operation: operation as unknown as TaskEntityOperation, getStoredState: () => stored }; +} + +describe("DurableEntityContext", () => { + it("exposes identity, operation name, input, and state accessors", () => { + const { operation, getStoredState } = createFakeOperation({ + name: "add", + input: 5, + initialState: 10, + }); + const df = new DurableEntityContext(operation); + + expect(df.operationName).toBe("add"); + expect(df.entityName).toBe("counter"); + expect(df.entityKey).toBe("user-1"); + expect(df.instanceId).toBe("@counter@user-1"); + expect(df.getInput()).toBe(5); + expect(df.getState()).toBe(10); + + df.setState(15); + expect(getStoredState()).toBe(15); + }); + + it("reports isNewlyConstructed based on whether prior state existed", () => { + const fresh = new DurableEntityContext(createFakeOperation({ name: "add" }).operation); + expect(fresh.isNewlyConstructed).toBe(true); + + const existing = new DurableEntityContext( + createFakeOperation({ name: "add", initialState: 1 }).operation, + ); + expect(existing.isNewlyConstructed).toBe(false); + }); + + it("exposes the entity id object and signals other entities through the core context", () => { + const { operation } = createFakeOperation({ name: "op", initialState: 1 }); + const df = new DurableEntityContext(operation); + + expect(df.entityId).toBe(operation.context.id); + expect(df.entityId.toString()).toBe("@counter@user-1"); + + const target = new EntityInstanceId("Other", "k-9"); + df.signalEntity(target, "poke", 42); + expect(operation.context.signalEntity).toHaveBeenCalledWith(target, "poke", 42); + }); + + it("returns the initializer default when there is no state", () => { + const { operation } = createFakeOperation({ name: "get" }); + const df = new DurableEntityContext(operation); + + expect(df.getState(() => 0)).toBe(0); + }); + + it("destructOnExit clears the state", () => { + const { operation, getStoredState } = createFakeOperation({ name: "delete", initialState: 42 }); + const df = new DurableEntityContext(operation); + + df.destructOnExit(); + expect(getStoredState()).toBeUndefined(); + }); + + it("resolveResult prefers an explicitly set result over the fallback", () => { + const { operation } = createFakeOperation({ name: "op" }); + const df = new DurableEntityContext(operation); + + expect(df.resolveResult("fallback")).toBe("fallback"); + df.return("explicit"); + expect(df.resolveResult("fallback")).toBe("explicit"); + }); +}); + +describe("wrapEntity", () => { + it("returns a zero-argument core factory unchanged", () => { + const factory: EntityFactory = () => ({ run: jest.fn() }); + expect(wrapEntity(factory)).toBe(factory); + }); + + it("wraps a classic entity function and dispatches operations through context.df", async () => { + const classic = (context: ClassicEntityContext) => { + const current = context.df.getState(() => 0) ?? 0; + const next = current + (context.df.getInput() ?? 0); + context.df.setState(next); + return next; + }; + + const factory = wrapEntity(classic); + expect(factory).not.toBe(classic); + + const { operation, getStoredState } = createFakeOperation({ + name: "add", + input: 7, + initialState: 3, + }); + const result = await factory().run(operation); + + expect(result).toBe(10); + expect(getStoredState()).toBe(10); + }); + + it("uses an explicitly set result over the function return value", async () => { + const classic = (context: ClassicEntityContext) => { + context.df.return("explicit-result"); + return "ignored-return"; + }; + + const factory = wrapEntity(classic); + const { operation } = createFakeOperation({ name: "op" }); + + await expect(factory().run(operation)).resolves.toBe("explicit-result"); + }); +}); diff --git a/packages/azure-functions-durable/test/unit/entity-id.spec.ts b/packages/azure-functions-durable/test/unit/entity-id.spec.ts new file mode 100644 index 0000000..5b64040 --- /dev/null +++ b/packages/azure-functions-durable/test/unit/entity-id.spec.ts @@ -0,0 +1,19 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { EntityInstanceId } from "@microsoft/durabletask-js"; +import { EntityId } from "../../src/entity-id"; + +describe("EntityId", () => { + it("constructs a classic v3 entity id, lowercasing the name and preserving the key", () => { + const id = new EntityId("Counter", "User-1"); + + expect(id.name).toBe("counter"); + expect(id.key).toBe("User-1"); + expect(id.toString()).toBe("@counter@User-1"); + }); + + it("is an EntityInstanceId so it works with the core entity APIs", () => { + expect(new EntityId("Counter", "k")).toBeInstanceOf(EntityInstanceId); + }); +}); diff --git a/packages/azure-functions-durable/test/unit/get-client.spec.ts b/packages/azure-functions-durable/test/unit/get-client.spec.ts new file mode 100644 index 0000000..9dcfa01 --- /dev/null +++ b/packages/azure-functions-durable/test/unit/get-client.spec.ts @@ -0,0 +1,60 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { InvocationContext } from "@azure/functions"; +import { TaskHubGrpcClient } from "@microsoft/durabletask-js"; +import { DurableFunctionsClient } from "../../src/client"; +import { getClient } from "../../src/get-client"; +import * as input from "../../src/input"; + +const CLIENT_CONFIG = { + taskHubName: "functions-taskhub", + rpcBaseUrl: "http://127.0.0.1:4711/rpc", + requiredQueryStringParameters: "code=secret&taskHub=functions-taskhub", +}; + +describe("getClient", () => { + it("builds a DurableFunctionsClient from the durableClient binding value", async () => { + const clientInput = input.durableClient(); + const context = new InvocationContext({ options: { extraInputs: [clientInput] } }); + context.extraInputs.set(clientInput, JSON.stringify(CLIENT_CONFIG)); + + const client = getClient(context); + + try { + expect(client).toBeInstanceOf(DurableFunctionsClient); + expect(client).toBeInstanceOf(TaskHubGrpcClient); + expect(client.taskHubName).toBe("functions-taskhub"); + } finally { + await client.stop(); + } + }); + + it("accepts an already-parsed binding object", async () => { + const clientInput = input.durableClient(); + const context = new InvocationContext({ options: { extraInputs: [clientInput] } }); + context.extraInputs.set(clientInput, CLIENT_CONFIG); + + const client = getClient(context); + + try { + expect(client).toBeInstanceOf(DurableFunctionsClient); + } finally { + await client.stop(); + } + }); + + it("throws when no durable client input binding is registered", () => { + const context = new InvocationContext({ options: { extraInputs: [] } }); + + expect(() => getClient(context)).toThrow(/durable client input binding/i); + }); + + it("throws when the binding value is not a valid client input", () => { + const clientInput = input.durableClient(); + const context = new InvocationContext({ options: { extraInputs: [clientInput] } }); + context.extraInputs.set(clientInput, 42); + + expect(() => getClient(context)).toThrow(/not a valid durable client input/i); + }); +}); diff --git a/packages/azure-functions-durable/test/unit/orchestration-context.spec.ts b/packages/azure-functions-durable/test/unit/orchestration-context.spec.ts new file mode 100644 index 0000000..3d1df67 --- /dev/null +++ b/packages/azure-functions-durable/test/unit/orchestration-context.spec.ts @@ -0,0 +1,235 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import * as durabletask from "@microsoft/durabletask-js"; +import { + EntityInstanceId, + OrchestrationContext, + RetryPolicy, + Task, + TOrchestrator, +} from "@microsoft/durabletask-js"; +import { + ClassicOrchestrationContext, + DurableOrchestrationContext, + wrapOrchestrator, +} from "../../src/orchestration-context"; +import { RetryOptions } from "../../src/retry-options"; + +/** Builds a fake core OrchestrationContext whose methods return sentinel values via jest mocks. */ +function createFakeCoreContext() { + const entities = { + callEntity: jest.fn().mockReturnValue("callEntity-task"), + signalEntity: jest.fn(), + }; + const replaySafeLogger = { + info: jest.fn(), + warn: jest.fn(), + error: jest.fn(), + debug: jest.fn(), + }; + const ctx = { + instanceId: "instance-1", + isReplaying: true, + currentUtcDateTime: new Date("2026-01-02T03:04:05.000Z"), + version: "1.2.3", + callActivity: jest.fn().mockReturnValue("callActivity-task"), + callSubOrchestrator: jest.fn().mockReturnValue("callSub-task"), + createTimer: jest.fn().mockReturnValue("timer-task"), + waitForExternalEvent: jest.fn().mockReturnValue("event-task"), + continueAsNew: jest.fn(), + setCustomStatus: jest.fn(), + sendEvent: jest.fn(), + newGuid: jest.fn().mockReturnValue("guid-1"), + createReplaySafeLogger: jest.fn(() => replaySafeLogger), + entities, + }; + return { ctx: ctx as unknown as OrchestrationContext, raw: ctx, entities, replaySafeLogger }; +} + +describe("DurableOrchestrationContext", () => { + it("exposes identity/replay properties and getInput from the core context", () => { + const { ctx } = createFakeCoreContext(); + const df = new DurableOrchestrationContext(ctx, { city: "Tokyo" }); + + expect(df.instanceId).toBe("instance-1"); + expect(df.isReplaying).toBe(true); + expect(df.currentUtcDateTime).toEqual(new Date("2026-01-02T03:04:05.000Z")); + expect(df.version).toBe("1.2.3"); + expect(df.getInput<{ city: string }>()).toEqual({ city: "Tokyo" }); + }); + + it("exposes parentInstanceId from the core context (undefined at top level)", () => { + const { ctx } = createFakeCoreContext(); + expect(new DurableOrchestrationContext(ctx, undefined).parentInstanceId).toBeUndefined(); + + const childCtx = { + parent: { name: "p", instanceId: "parent-1", taskScheduledId: 1 }, + } as unknown as OrchestrationContext; + expect(new DurableOrchestrationContext(childCtx, undefined).parentInstanceId).toBe("parent-1"); + }); + + it("tracks the custom status locally and forwards it to the core context", () => { + const { ctx, raw } = createFakeCoreContext(); + const df = new DurableOrchestrationContext(ctx, undefined); + + expect(df.customStatus).toBeUndefined(); + df.setCustomStatus({ stage: "processing" }); + expect(df.customStatus).toEqual({ stage: "processing" }); + expect(raw.setCustomStatus).toHaveBeenCalledWith({ stage: "processing" }); + }); + + it("forwards callActivity and callActivityWithRetry to the core context", () => { + const { ctx, raw } = createFakeCoreContext(); + const df = new DurableOrchestrationContext(ctx, undefined); + + expect(df.callActivity("sayHello", "Tokyo")).toBe("callActivity-task"); + expect(raw.callActivity).toHaveBeenCalledWith("sayHello", "Tokyo"); + + const retry = new RetryOptions(1000, 3); + df.callActivityWithRetry("flaky", retry, "input"); + const retryCall = raw.callActivity.mock.calls[1]; + expect(retryCall[0]).toBe("flaky"); + expect(retryCall[1]).toBe("input"); + expect(retryCall[2].retry).toBeInstanceOf(RetryPolicy); + expect(retryCall[2].retry.maxNumberOfAttempts).toBe(3); + expect(retryCall[2].retry.firstRetryIntervalInMilliseconds).toBe(1000); + }); + + it("forwards sub-orchestration calls with instanceId and retry options", () => { + const { ctx, raw } = createFakeCoreContext(); + const df = new DurableOrchestrationContext(ctx, undefined); + + df.callSubOrchestrator("child", "in", "child-id"); + expect(raw.callSubOrchestrator).toHaveBeenCalledWith("child", "in", { instanceId: "child-id" }); + + df.callSubOrchestrator("child2", "in2"); + expect(raw.callSubOrchestrator).toHaveBeenLastCalledWith("child2", "in2", undefined); + + const retry = new RetryOptions(1000, 2); + df.callSubOrchestratorWithRetry("child3", retry, "in3", "id3"); + const subCalls = raw.callSubOrchestrator.mock.calls; + const lastSub = subCalls[subCalls.length - 1]; + expect(lastSub[0]).toBe("child3"); + expect(lastSub[1]).toBe("in3"); + expect(lastSub[2].instanceId).toBe("id3"); + expect(lastSub[2].retry).toBeInstanceOf(RetryPolicy); + expect(lastSub[2].retry.maxNumberOfAttempts).toBe(2); + }); + + it("forwards timers, events, status, and entity operations", () => { + const { ctx, raw, entities } = createFakeCoreContext(); + const df = new DurableOrchestrationContext(ctx, undefined); + const entityId = new EntityInstanceId("Counter", "k1"); + + df.createTimer(123); + expect(raw.createTimer).toHaveBeenCalledWith(123); + + df.waitForExternalEvent("Approval"); + expect(raw.waitForExternalEvent).toHaveBeenCalledWith("Approval"); + + df.continueAsNew("next"); + expect(raw.continueAsNew).toHaveBeenCalledWith("next", true); + + df.setCustomStatus({ stage: "x" }); + expect(raw.setCustomStatus).toHaveBeenCalledWith({ stage: "x" }); + + expect(df.newGuid("ignored-instance-id")).toBe("guid-1"); + expect(raw.newGuid).toHaveBeenCalledTimes(1); + + df.callEntity(entityId, "add", 5); + expect(entities.callEntity).toHaveBeenCalledWith(entityId, "add", 5); + + df.signalEntity(entityId, "reset"); + expect(entities.signalEntity).toHaveBeenCalledWith(entityId, "reset", undefined); + }); + + it("throws for callHttp, which has no durabletask equivalent", () => { + const { ctx } = createFakeCoreContext(); + const df = new DurableOrchestrationContext(ctx, undefined); + + expect(() => df.callHttp({ method: "GET", uri: "https://example.test" })).toThrow(/callHttp/); + }); + + it("exposes Task.all / Task.any that forward to the core combinators", () => { + const { ctx } = createFakeCoreContext(); + const df = new DurableOrchestrationContext(ctx, undefined); + const tasks = ["t1", "t2"] as unknown as Task[]; + + const whenAllSpy = jest.spyOn(durabletask, "whenAll").mockReturnValue("all-result" as never); + const whenAnySpy = jest.spyOn(durabletask, "whenAny").mockReturnValue("any-result" as never); + try { + expect(df.Task.all(tasks)).toBe("all-result"); + expect(whenAllSpy).toHaveBeenCalledWith(tasks); + expect(df.Task.any(tasks)).toBe("any-result"); + expect(whenAnySpy).toHaveBeenCalledWith(tasks); + } finally { + whenAllSpy.mockRestore(); + whenAnySpy.mockRestore(); + } + }); +}); + +describe("wrapOrchestrator", () => { + it("returns a two-parameter core-native orchestrator unchanged", () => { + // A core-native orchestrator declares (ctx, input); a non-generator that returns a value is a + // valid TOrchestrator and keeps the arity at 2 so it passes through unwrapped. + const native: TOrchestrator = (_ctx, _input) => "native"; + expect(wrapOrchestrator(native)).toBe(native); + }); + + it("wraps a single-parameter classic orchestrator and drives it through context.df", async () => { + const classic = function* ( + context: ClassicOrchestrationContext, + ): Generator, string, unknown> { + const input = context.df.getInput(); + const first = (yield context.df.callActivity("a", input)) as string; + return `done:${first}`; + }; + + const wrapped = wrapOrchestrator(classic); + // The wrapper is core-native shape (ctx, input). + expect(wrapped).not.toBe(classic); + + const { ctx, raw } = createFakeCoreContext(); + // The engine gates on Symbol.asyncIterator: the wrapper MUST return an async generator or the + // core executor never drives it (it would complete immediately with the raw return value). + const gen = wrapped(ctx, "INPUT") as AsyncGenerator; + expect(typeof (gen as { [Symbol.asyncIterator]?: unknown })[Symbol.asyncIterator]).toBe( + "function", + ); + + const firstYield = await gen.next(); + expect(raw.callActivity).toHaveBeenCalledWith("a", "INPUT"); + expect(firstYield.value).toBe("callActivity-task"); + expect(firstYield.done).toBe(false); + + const final = await gen.next("ACTIVITY_RESULT"); + expect(final.done).toBe(true); + expect(final.value).toBe("done:ACTIVITY_RESULT"); + }); + + it("exposes a replay-safe logger as context.log/error on the classic context", async () => { + // A classic orchestrator may be a plain (non-generator) function that returns a value; the + // wrapper still invokes it with the full classic context, so log wiring is exercised here. + const classic = (context: ClassicOrchestrationContext): string => { + context.log("hi"); + context.error("boom"); + return "logged"; + }; + + const { ctx, replaySafeLogger } = createFakeCoreContext(); + const wrapped = wrapOrchestrator(classic); + const gen = wrapped(ctx, undefined) as AsyncGenerator; + + const done = await gen.next(); + expect(done.done).toBe(true); + expect(done.value).toBe("logged"); + // The wrapper builds the classic context's log methods from the CORE replay-safe logger. + expect( + (ctx as unknown as { createReplaySafeLogger: jest.Mock }).createReplaySafeLogger, + ).toHaveBeenCalledTimes(1); + expect(replaySafeLogger.info).toHaveBeenCalledWith("hi"); + expect(replaySafeLogger.error).toHaveBeenCalledWith("boom"); + }); +}); diff --git a/packages/azure-functions-durable/test/unit/query-types.spec.ts b/packages/azure-functions-durable/test/unit/query-types.spec.ts new file mode 100644 index 0000000..40718e2 --- /dev/null +++ b/packages/azure-functions-durable/test/unit/query-types.spec.ts @@ -0,0 +1,100 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { OrchestrationState, OrchestrationStatus } from "@microsoft/durabletask-js"; +import { EntityStateResponse } from "../../src/entity-state-response"; +import { + DurableOrchestrationStatus, + OrchestrationRuntimeStatus, + toDurableOrchestrationStatus, + toOrchestrationRuntimeStatus, +} from "../../src/orchestration-status"; +import { PurgeHistoryResult } from "../../src/purge-history-result"; + +describe("toOrchestrationRuntimeStatus", () => { + it("maps core statuses to the classic v3 runtime status strings", () => { + expect(toOrchestrationRuntimeStatus(OrchestrationStatus.RUNNING)).toBe( + OrchestrationRuntimeStatus.Running, + ); + expect(toOrchestrationRuntimeStatus(OrchestrationStatus.COMPLETED)).toBe( + OrchestrationRuntimeStatus.Completed, + ); + expect(toOrchestrationRuntimeStatus(OrchestrationStatus.CONTINUED_AS_NEW)).toBe( + OrchestrationRuntimeStatus.ContinuedAsNew, + ); + expect(toOrchestrationRuntimeStatus(OrchestrationStatus.FAILED)).toBe( + OrchestrationRuntimeStatus.Failed, + ); + expect(toOrchestrationRuntimeStatus(OrchestrationStatus.TERMINATED)).toBe( + OrchestrationRuntimeStatus.Terminated, + ); + expect(toOrchestrationRuntimeStatus(OrchestrationStatus.PENDING)).toBe( + OrchestrationRuntimeStatus.Pending, + ); + expect(toOrchestrationRuntimeStatus(OrchestrationStatus.SUSPENDED)).toBe( + OrchestrationRuntimeStatus.Suspended, + ); + }); +}); + +describe("toDurableOrchestrationStatus", () => { + it("maps a core OrchestrationState and deserializes JSON payloads", () => { + const created = new Date("2026-01-01T00:00:00.000Z"); + const updated = new Date("2026-01-02T00:00:00.000Z"); + const state = new OrchestrationState( + "instance-1", + "helloOrchestrator", + OrchestrationStatus.COMPLETED, + created, + updated, + JSON.stringify({ city: "Tokyo" }), + JSON.stringify("Hello Tokyo"), + JSON.stringify({ stage: "done" }), + ); + + const status = toDurableOrchestrationStatus(state); + + expect(status).toBeInstanceOf(DurableOrchestrationStatus); + expect(status.name).toBe("helloOrchestrator"); + expect(status.instanceId).toBe("instance-1"); + expect(status.createdTime).toBe(created); + expect(status.lastUpdatedTime).toBe(updated); + expect(status.input).toEqual({ city: "Tokyo" }); + expect(status.output).toBe("Hello Tokyo"); + expect(status.customStatus).toEqual({ stage: "done" }); + expect(status.runtimeStatus).toBe(OrchestrationRuntimeStatus.Completed); + }); + + it("leaves payloads undefined when the core state omits them", () => { + const state = new OrchestrationState( + "instance-2", + "orch", + OrchestrationStatus.RUNNING, + new Date(), + new Date(), + ); + + const status = toDurableOrchestrationStatus(state); + + expect(status.input).toBeUndefined(); + expect(status.output).toBeUndefined(); + expect(status.customStatus).toBeUndefined(); + expect(status.runtimeStatus).toBe(OrchestrationRuntimeStatus.Running); + }); +}); + +describe("EntityStateResponse / PurgeHistoryResult", () => { + it("EntityStateResponse carries existence and state", () => { + const present = new EntityStateResponse(true, 42); + expect(present.entityExists).toBe(true); + expect(present.entityState).toBe(42); + + const absent = new EntityStateResponse(false); + expect(absent.entityExists).toBe(false); + expect(absent.entityState).toBeUndefined(); + }); + + it("PurgeHistoryResult carries the deleted count", () => { + expect(new PurgeHistoryResult(3).instancesDeleted).toBe(3); + }); +}); diff --git a/packages/azure-functions-durable/test/unit/retry-options.spec.ts b/packages/azure-functions-durable/test/unit/retry-options.spec.ts new file mode 100644 index 0000000..f55793e --- /dev/null +++ b/packages/azure-functions-durable/test/unit/retry-options.spec.ts @@ -0,0 +1,31 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { RetryPolicy } from "@microsoft/durabletask-js"; +import { RetryOptions } from "../../src/retry-options"; + +describe("RetryOptions", () => { + it("maps the classic (firstRetryInterval, maxAttempts) shape onto a core RetryPolicy", () => { + const options = new RetryOptions(5000, 3); + + const policy = options.toRetryPolicy(); + + expect(policy).toBeInstanceOf(RetryPolicy); + expect(policy.firstRetryIntervalInMilliseconds).toBe(5000); + expect(policy.maxNumberOfAttempts).toBe(3); + expect(policy.backoffCoefficient).toBe(1); + }); + + it("forwards backoff and interval overrides to the core RetryPolicy", () => { + const options = new RetryOptions(1000, 5); + options.backoffCoefficient = 2; + options.maxRetryIntervalInMilliseconds = 60000; + options.retryTimeoutInMilliseconds = 300000; + + const policy = options.toRetryPolicy(); + + expect(policy.backoffCoefficient).toBe(2); + expect(policy.maxRetryIntervalInMilliseconds).toBe(60000); + expect(policy.retryTimeoutInMilliseconds).toBe(300000); + }); +}); diff --git a/packages/azure-functions-durable/test/unit/trigger.spec.ts b/packages/azure-functions-durable/test/unit/trigger.spec.ts new file mode 100644 index 0000000..882337c --- /dev/null +++ b/packages/azure-functions-durable/test/unit/trigger.spec.ts @@ -0,0 +1,39 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import * as input from "../../src/input"; +import * as trigger from "../../src/trigger"; + +function flag(binding: Record): unknown { + return binding.durableRequiresGrpc; +} + +describe("durable triggers and inputs", () => { + it("orchestration trigger uses the extension type string and opts in to gRPC", () => { + const orchestration = trigger.orchestration(); + + expect(orchestration.type).toBe("orchestrationTrigger"); + expect(flag(orchestration)).toBe(true); + }); + + it("activity trigger uses the extension type string and opts in to gRPC", () => { + const activity = trigger.activity(); + + expect(activity.type).toBe("activityTrigger"); + expect(flag(activity)).toBe(true); + }); + + it("entity trigger uses the extension type string and opts in to gRPC", () => { + const entity = trigger.entity(); + + expect(entity.type).toBe("entityTrigger"); + expect(flag(entity)).toBe(true); + }); + + it("durableClient input uses the extension type string and opts in to gRPC", () => { + const clientInput = input.durableClient(); + + expect(clientInput.type).toBe("durableClient"); + expect(flag(clientInput)).toBe(true); + }); +}); diff --git a/packages/azure-functions-durable/test/unit/worker.spec.ts b/packages/azure-functions-durable/test/unit/worker.spec.ts new file mode 100644 index 0000000..7e32302 --- /dev/null +++ b/packages/azure-functions-durable/test/unit/worker.spec.ts @@ -0,0 +1,48 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { DurableFunctionsWorker } from "../../src/worker"; + +describe("DurableFunctionsWorker", () => { + it("decodes base64, delegates to processOrchestratorRequest, and re-encodes the response", async () => { + const worker = new DurableFunctionsWorker(); + const responseBytes = Buffer.from("orchestrator response"); + const processOrchestratorRequest = jest + .spyOn(worker, "processOrchestratorRequest") + .mockResolvedValue(responseBytes); + + const actual = await worker.handleOrchestratorRequest( + Buffer.from("orchestrator request").toString("base64"), + ); + + expect(actual).toBe(responseBytes.toString("base64")); + expect(processOrchestratorRequest).toHaveBeenCalledTimes(1); + expect(Buffer.from(processOrchestratorRequest.mock.calls[0][0]).toString()).toBe( + "orchestrator request", + ); + }); + + it("decodes base64, delegates to processEntityBatchRequest, and re-encodes the response", async () => { + const worker = new DurableFunctionsWorker(); + const responseBytes = Buffer.from("entity batch response"); + const processEntityBatchRequest = jest + .spyOn(worker, "processEntityBatchRequest") + .mockResolvedValue(responseBytes); + + const actual = await worker.handleEntityBatchRequest( + Buffer.from("entity batch request").toString("base64"), + ); + + expect(actual).toBe(responseBytes.toString("base64")); + expect(processEntityBatchRequest).toHaveBeenCalledTimes(1); + expect(Buffer.from(processEntityBatchRequest.mock.calls[0][0]).toString()).toBe( + "entity batch request", + ); + }); + + it("rejects empty base64 requests", async () => { + const worker = new DurableFunctionsWorker(); + + await expect(worker.handleOrchestratorRequest("")).rejects.toThrow(TypeError); + }); +}); diff --git a/packages/azure-functions-durable/tsconfig.build.json b/packages/azure-functions-durable/tsconfig.build.json new file mode 100644 index 0000000..f891744 --- /dev/null +++ b/packages/azure-functions-durable/tsconfig.build.json @@ -0,0 +1,9 @@ +{ + "extends": "./tsconfig.json", + "compilerOptions": { + "baseUrl": ".", + "paths": { + "@microsoft/durabletask-js": ["../durabletask-js/dist/index"] + } + } +} diff --git a/packages/azure-functions-durable/tsconfig.json b/packages/azure-functions-durable/tsconfig.json new file mode 100644 index 0000000..0180107 --- /dev/null +++ b/packages/azure-functions-durable/tsconfig.json @@ -0,0 +1,9 @@ +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { + "rootDir": "./src", + "outDir": "./dist" + }, + "include": ["src"], + "exclude": ["node_modules", "dist", "test"] +} diff --git a/packages/durabletask-js/src/proto/orchestrator_service_pb.d.ts b/packages/durabletask-js/src/proto/orchestrator_service_pb.d.ts index 1505288..f3076db 100644 --- a/packages/durabletask-js/src/proto/orchestrator_service_pb.d.ts +++ b/packages/durabletask-js/src/proto/orchestrator_service_pb.d.ts @@ -1756,8 +1756,8 @@ export class OrchestratorRequest extends jspb.Message { static extensions: {[key: number]: jspb.ExtensionFieldInfo}; static extensionsBinary: {[key: number]: jspb.ExtensionFieldBinaryInfo}; static serializeBinaryToWriter(message: OrchestratorRequest, writer: jspb.BinaryWriter): void; - static deserializeBinary(bytes: Uint8Array): OrchestratorRequest; static deserializeBinaryFromReader(message: OrchestratorRequest, reader: jspb.BinaryReader): OrchestratorRequest; + static deserializeBinary(bytes: Uint8Array): OrchestratorRequest; } export namespace OrchestratorRequest { 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 9fec6f3..e4f43f4 100644 --- a/packages/durabletask-js/src/worker/task-hub-grpc-worker.ts +++ b/packages/durabletask-js/src/worker/task-hub-grpc-worker.ts @@ -313,6 +313,58 @@ export class TaskHubGrpcWorker { return name.toLowerCase(); } + /** + * Processes a single serialized TaskHubSidecarService OrchestratorRequest and + * returns the serialized OrchestratorResponse. + * + * @param request - The protobuf-encoded OrchestratorRequest bytes. + * @returns The protobuf-encoded OrchestratorResponse bytes. + * + * @remarks + * This is intended for host integrations, such as Azure Functions, that drive a + * single orchestration work item per invocation instead of running the + * long-lived gRPC worker loop. It reuses the same execution path as the worker + * loop, capturing the response in-process rather than completing it over gRPC. + * Host integrations own any transport-specific encoding (for example base64). + */ + async processOrchestratorRequest(request: Uint8Array): Promise { + const req = pb.OrchestratorRequest.deserializeBinary(request); + const stub = new CapturingSidecarStub(); + await this._executeOrchestratorInternal(req, "", stub as unknown as stubs.TaskHubSidecarServiceClient); + + if (!stub.orchestratorResponse) { + throw new Error("Orchestrator execution did not produce a response."); + } + + return stub.orchestratorResponse.serializeBinary(); + } + + /** + * Processes a single serialized TaskHubSidecarService EntityBatchRequest and + * returns the serialized EntityBatchResult. + * + * @param request - The protobuf-encoded EntityBatchRequest bytes. + * @returns The protobuf-encoded EntityBatchResult bytes. + * + * @remarks + * This is intended for host integrations, such as Azure Functions, that drive a + * single entity batch work item per invocation instead of running the + * long-lived gRPC worker loop. It reuses the same execution path as the worker + * loop, capturing the result in-process rather than completing it over gRPC. + * Host integrations own any transport-specific encoding (for example base64). + */ + async processEntityBatchRequest(request: Uint8Array): Promise { + const req = pb.EntityBatchRequest.deserializeBinary(request); + const stub = new CapturingSidecarStub(); + await this._executeEntityInternal(req, "", stub as unknown as stubs.TaskHubSidecarServiceClient); + + if (!stub.entityResult) { + throw new Error("Entity batch execution did not produce a result."); + } + + return stub.entityResult.serializeBinary(); + } + /** * In node.js we don't require a new thread as we have a main event loop * Therefore, we open the stream and simply listen through the eventemitter behind the scenes @@ -1157,3 +1209,47 @@ export class TaskHubGrpcWorker { } } } + +/** + * A minimal in-process stand-in for the TaskHubSidecarService client that captures the + * completion payload instead of sending it over gRPC. + * + * @remarks + * This lets host integrations reuse the worker's existing execution path for a single work + * item (see {@link TaskHubGrpcWorker.processOrchestratorRequest} and + * {@link TaskHubGrpcWorker.processEntityBatchRequest}) without opening a gRPC channel. Only the + * completion/abandon methods used by those execution paths are implemented. + */ +class CapturingSidecarStub { + orchestratorResponse?: pb.OrchestratorResponse; + entityResult?: pb.EntityBatchResult; + + completeOrchestratorTask( + request: pb.OrchestratorResponse, + _metadata: grpc.Metadata, + callback: (error: grpc.ServiceError | null, response: Empty) => void, + ): void { + this.orchestratorResponse = request; + callback(null, new Empty()); + } + + completeEntityTask( + request: pb.EntityBatchResult, + _metadata: grpc.Metadata, + callback: (error: grpc.ServiceError | null, response: Empty) => void, + ): void { + this.entityResult = request; + callback(null, new Empty()); + } + + abandonTaskOrchestratorWorkItem( + _request: pb.AbandonOrchestrationTaskRequest, + _metadata: grpc.Metadata, + callback: (error: grpc.ServiceError | null, response: Empty) => void, + ): void { + // Abandon is a no-op for the single-work-item host path: the version-mismatch abandon branch + // in _executeOrchestratorInternal calls this, but processOrchestratorRequest only surfaces a + // completion response. Matches the Python provider, whose null stub no-ops abandon. + callback(null, new Empty()); + } +} diff --git a/packages/durabletask-js/test/functions-grpc-support.spec.ts b/packages/durabletask-js/test/functions-grpc-support.spec.ts new file mode 100644 index 0000000..21f6d75 --- /dev/null +++ b/packages/durabletask-js/test/functions-grpc-support.spec.ts @@ -0,0 +1,68 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { OrchestrationContext, TaskEntity, TaskHubGrpcWorker, TOrchestrator } from "../src"; +import * as pb from "../src/proto/orchestrator_service_pb"; +import { newExecutionStartedEvent, newOrchestratorStartedEvent } from "../src/utils/pb-helper.util"; +import { NoOpLogger } from "../src/types/logger.type"; + +const TEST_INSTANCE_ID = "functions-grpc-instance"; + +class CounterEntity extends TaskEntity { + increment(): number { + this.state++; + return this.state; + } + + protected initializeState(): number { + return 0; + } +} + +describe("Functions gRPC support surface", () => { + it("processes a single serialized orchestration request without the gRPC worker loop", async () => { + const worker = new TaskHubGrpcWorker({ logger: new NoOpLogger() }); + const orchestrator: TOrchestrator = async (_ctx: OrchestrationContext) => "done"; + const name = worker.addOrchestrator(orchestrator); + + const request = new pb.OrchestratorRequest(); + request.setInstanceid(TEST_INSTANCE_ID); + request.setNeweventsList([ + newOrchestratorStartedEvent(new Date("2026-01-01T00:00:00.000Z")), + newExecutionStartedEvent(name, TEST_INSTANCE_ID, undefined), + ]); + + const responseBytes = await worker.processOrchestratorRequest(request.serializeBinary()); + const response = pb.OrchestratorResponse.deserializeBinary(responseBytes); + + expect(response.getInstanceid()).toBe(TEST_INSTANCE_ID); + const completed = response.getActionsList()[0].getCompleteorchestration(); + expect(completed?.getOrchestrationstatus()).toBe(pb.OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED); + expect(completed?.getResult()?.getValue()).toBe('"done"'); + }); + + it("processes a single serialized entity batch request without the gRPC worker loop", async () => { + const worker = new TaskHubGrpcWorker({ logger: new NoOpLogger() }); + worker.addNamedEntity("counter", () => new CounterEntity()); + + const request = createEntityBatchRequest("counter", "key1"); + const responseBytes = await worker.processEntityBatchRequest(request.serializeBinary()); + const response = pb.EntityBatchResult.deserializeBinary(responseBytes); + + expect(response.getResultsList()).toHaveLength(1); + expect(response.getResultsList()[0].getSuccess()?.getResult()?.getValue()).toBe("1"); + expect(response.getEntitystate()?.getValue()).toBe("1"); + }); +}); + +function createEntityBatchRequest(entityName: string, entityKey: string): pb.EntityBatchRequest { + const request = new pb.EntityBatchRequest(); + request.setInstanceid(`@${entityName}@${entityKey}`); + + const operation = new pb.OperationRequest(); + operation.setOperation("increment"); + operation.setRequestid("req-1"); + request.setOperationsList([operation]); + + return request; +}