From ff79d4f7474c2dbef3e2f4b6f27d730a9df0f410 Mon Sep 17 00:00:00 2001 From: wangbill Date: Wed, 5 Aug 2026 11:28:38 -0700 Subject: [PATCH 1/4] Add worker-level middleware Introduce typed orchestration and activity middleware pipelines across gRPC, test, and Azure-managed worker construction paths. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 5 +- README.md | 90 ++- .../durabletask-js-azuremanaged/CHANGELOG.md | 2 + .../durabletask-js-azuremanaged/README.md | 9 + .../src/worker-builder.ts | 28 + .../test/unit/worker-builder.spec.ts | 24 +- packages/durabletask-js/src/index.ts | 11 + .../src/testing/in-memory-backend.ts | 32 +- .../durabletask-js/src/testing/test-worker.ts | 35 +- .../src/worker/activity-executor.ts | 39 +- .../durabletask-js/src/worker/middleware.ts | 366 ++++++++++++ .../src/worker/orchestration-executor.ts | 161 ++++-- .../worker/runtime-orchestration-context.ts | 97 +--- .../src/worker/task-hub-grpc-worker.ts | 86 ++- .../test/functions-grpc-support.spec.ts | 59 ++ .../durabletask-js/test/middleware.spec.ts | 540 ++++++++++++++++++ .../test/test-worker-middleware.spec.ts | 63 ++ 17 files changed, 1477 insertions(+), 170 deletions(-) create mode 100644 packages/durabletask-js/src/worker/middleware.ts create mode 100644 packages/durabletask-js/test/middleware.spec.ts create mode 100644 packages/durabletask-js/test/test-worker-middleware.spec.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index e2f8a3e8..cf45847b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,8 +2,9 @@ ### New -### Fixes +- Add first-class worker-level orchestration and activity middleware with typed contexts, nonserialized host features, validation, and in-process host integration. +### Fixes ## v0.4.0 (2026-07-31) @@ -70,7 +71,7 @@ - fix: clear customStatus on continue-as-new in InMemoryOrchestrationBackend ([#155](https://github.com/microsoft/durabletask-js/pull/155)) - fix: propagate parent notification from composite tasks (WhenAllTask/WhenAnyTask) ([#150](https://github.com/microsoft/durabletask-js/pull/150)) - fix: use deterministic time in createTimer instead of Date.now() ([#146](https://github.com/microsoft/durabletask-js/pull/146)) -- Fix WhenAllTask constructor resetting _completedTasks counter ([#143](https://github.com/microsoft/durabletask-js/pull/143)) +- Fix WhenAllTask constructor resetting \_completedTasks counter ([#143](https://github.com/microsoft/durabletask-js/pull/143)) - Fix retry handler treating undefined/null/NaN/Infinity as retry signal ([#142](https://github.com/microsoft/durabletask-js/pull/142)) - Release v0.3.0 ([#147](https://github.com/microsoft/durabletask-js/pull/147)) diff --git a/README.md b/README.md index 683638b2..879d50b8 100644 --- a/README.md +++ b/README.md @@ -18,18 +18,86 @@ worker.addActivity(myActivity); worker.addEntity(myEntity); const orchestrationResponseBytes = await worker.processOrchestratorRequest(orchestrationRequestBytes); +const activityResponseBytes = await worker.processActivityRequest(activityRequestBytes); 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`. +## Durable Task middleware + +Worker-level middleware runs around orchestration or activity execution. The first registered middleware is outermost, so middleware runs in registration order before user code and unwinds in reverse order afterward. + +```typescript +import { ActivityMiddleware, OrchestrationMiddleware, TaskHubGrpcWorker } from "@microsoft/durabletask-js"; + +const orchestrationLogging: OrchestrationMiddleware = async (context, next) => { + const logger = context.orchestrationContext.createReplaySafeLogger(appLogger); + logger.info(`Starting ${context.name} (${context.instanceId})`); + + await next(context); + + if (context.failure) { + logger.error(`Failed ${context.name}: ${context.failure.message}`); + } else if (context.result !== undefined) { + logger.info(`Completed ${context.name}`); + } +}; + +const activityCache: ActivityMiddleware = async (context, next) => { + const cached = await cache.get(context.name, context.input); + if (cached !== undefined) { + context.setResult(cached); + return; + } + + await next(context); + await cache.set(context.name, context.input, context.result); +}; + +const worker = new TaskHubGrpcWorker(); +worker + .useOrchestrationMiddleware(orchestrationLogging) + .useActivityMiddleware(activityCache) + .addOrchestrator(myOrchestrator); +worker.addActivity(myActivity); +``` + +Orchestration middleware participates in generator replay: code before `next(context)` re-executes on each replay pass, while `next` remains suspended with the orchestrator and code after it unwinds only when the orchestration reaches a terminal result or failure. Orchestration middleware must call `next(context)` exactly once and must not return before it completes. It cannot short-circuit or replace the orchestration result. Activity middleware must either call `next(context)` once without returning before it completes, or explicitly short-circuit with `context.setResult(...)`; calling `next` after `setResult` is invalid, while calling `setResult` after `next` may replace a result. Missing, duplicate, and incomplete `next` calls are rejected. In normal `async` middleware, use `await next(context)` as shown above. + +Orchestration middleware follows the same determinism rules as orchestrator code. Do not await nondurable promises or use wall-clock time, random values, mutable process state, file/network I/O, or host state that can change across replay. Use durable context APIs and `createReplaySafeLogger()` instead. Put I/O in activities or activity middleware. + +### Host features + +`MiddlewareFeatures` is a typed, symbol-keyed map for per-invocation host objects. Feature values remain process-local and are never serialized into durable history. + +```typescript +import { MiddlewareFeatures, createMiddlewareFeature } from "@microsoft/durabletask-js"; + +interface HostInvocation { + invocationId: string; +} + +const hostInvocation = createMiddlewareFeature("host invocation"); +const features = new MiddlewareFeatures().set(hostInvocation, currentInvocation); + +worker.useActivityMiddleware(async (context, next) => { + const invocation = context.features.get(hostInvocation); + await next(context); +}); + +await worker.processActivityRequest(activityRequestBytes, features); +``` + +Hosts can pass features to `processOrchestratorRequest()` and `processActivityRequest()`. Durable Task middleware is for durable-aware invocation data such as task name, instance ID, version, parent, tags, input, replay state, result, and failure. Host or framework middleware remains the right layer for concerns that apply to every host invocation, such as HTTP headers, authentication preprocessing, and nondurable bindings. Entity middleware is not supported in this version. + ## npm packages The following npm packages are available for download. -| Name | Latest version | Description | -| - | - | - | -| Core SDK | [![npm version](https://img.shields.io/npm/v/@microsoft/durabletask-js)](https://www.npmjs.com/package/@microsoft/durabletask-js) | Core Durable Task SDK for JavaScript/TypeScript. | +| Name | Latest version | Description | +| ---------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| Core SDK | [![npm version](https://img.shields.io/npm/v/@microsoft/durabletask-js)](https://www.npmjs.com/package/@microsoft/durabletask-js) | Core Durable Task SDK for JavaScript/TypeScript. | | AzureManaged SDK | [![npm version](https://img.shields.io/npm/v/@microsoft/durabletask-js-azuremanaged)](https://www.npmjs.com/package/@microsoft/durabletask-js-azuremanaged) | Azure-managed [Durable Task Scheduler](https://learn.microsoft.com/azure/azure-functions/durable/durable-functions-task-scheduler) support for the Durable Task JavaScript SDK. | ## Prerequisites @@ -50,15 +118,8 @@ npm install @microsoft/durabletask-js @microsoft/durabletask-js-azuremanaged You can then use the following code to define a simple "Hello, cities" durable orchestration. ```typescript -import { - ActivityContext, - OrchestrationContext, - TOrchestrator, -} from "@microsoft/durabletask-js"; -import { - createAzureManagedClient, - createAzureManagedWorkerBuilder, -} from "@microsoft/durabletask-js-azuremanaged"; +import { ActivityContext, OrchestrationContext, TOrchestrator } from "@microsoft/durabletask-js"; +import { createAzureManagedClient, createAzureManagedWorkerBuilder } from "@microsoft/durabletask-js-azuremanaged"; // Define an activity function const sayHello = async (_: ActivityContext, name: string): Promise => { @@ -125,10 +186,7 @@ An orchestration can wait for external events, such as a human approval, with op ```typescript import { whenAny } from "@microsoft/durabletask-js"; -const purchaseOrderWorkflow: TOrchestrator = async function* ( - ctx: OrchestrationContext, - order: Order, -): any { +const purchaseOrderWorkflow: TOrchestrator = async function* (ctx: OrchestrationContext, order: Order): any { // Orders under $1000 are auto-approved if (order.cost < 1000) { return "Auto-approved"; diff --git a/packages/durabletask-js-azuremanaged/CHANGELOG.md b/packages/durabletask-js-azuremanaged/CHANGELOG.md index 532f5c30..384bb5a5 100644 --- a/packages/durabletask-js-azuremanaged/CHANGELOG.md +++ b/packages/durabletask-js-azuremanaged/CHANGELOG.md @@ -2,6 +2,8 @@ ### New +- Add orchestration and activity middleware registration to `DurableTaskAzureManagedWorkerBuilder`. + ### Fixes ## v0.4.0 (2026-07-31) diff --git a/packages/durabletask-js-azuremanaged/README.md b/packages/durabletask-js-azuremanaged/README.md index 029e569b..35067c49 100644 --- a/packages/durabletask-js-azuremanaged/README.md +++ b/packages/durabletask-js-azuremanaged/README.md @@ -26,6 +26,15 @@ const client = createAzureManagedClient( const worker = createAzureManagedWorkerBuilder( "Endpoint=https://myservice.durabletask.io;Authentication=DefaultAzure;TaskHub=myTaskHub", ) + .useOrchestrationMiddleware(async (context, next) => { + const logger = context.orchestrationContext.createReplaySafeLogger(appLogger); + logger.info(`Running ${context.name} (${context.instanceId})`); + await next(context); + }) + .useActivityMiddleware(async (context, next) => { + await next(context); + appLogger.info(`Activity ${context.name} returned`, context.result); + }) .addOrchestrator(myOrchestrator) .addActivity(myActivity) .build(); diff --git a/packages/durabletask-js-azuremanaged/src/worker-builder.ts b/packages/durabletask-js-azuremanaged/src/worker-builder.ts index f749ea79..7d85c9d5 100644 --- a/packages/durabletask-js-azuremanaged/src/worker-builder.ts +++ b/packages/durabletask-js-azuremanaged/src/worker-builder.ts @@ -15,6 +15,8 @@ import { ConsoleLogger, VersioningOptions, WorkItemFilters, + ActivityMiddleware, + OrchestrationMiddleware, } from "@microsoft/durabletask-js"; /** @@ -31,6 +33,8 @@ export class DurableTaskAzureManagedWorkerBuilder { private _shutdownTimeoutMs?: number; private _versioning?: VersioningOptions; private _workItemFilters?: WorkItemFilters | "auto"; + private _orchestrationMiddleware: OrchestrationMiddleware[] = []; + private _activityMiddleware: ActivityMiddleware[] = []; /** * Creates a new instance of DurableTaskAzureManagedWorkerBuilder. @@ -187,6 +191,22 @@ export class DurableTaskAzureManagedWorkerBuilder { return this; } + /** + * Adds orchestration middleware to the worker. + */ + useOrchestrationMiddleware(middleware: OrchestrationMiddleware): DurableTaskAzureManagedWorkerBuilder { + this._orchestrationMiddleware.push(middleware); + return this; + } + + /** + * Adds activity middleware to the worker. + */ + useActivityMiddleware(middleware: ActivityMiddleware): DurableTaskAzureManagedWorkerBuilder { + this._activityMiddleware.push(middleware); + return this; + } + /** * Registers an entity factory with the worker. * The entity name is derived from the factory function name. @@ -297,6 +317,14 @@ export class DurableTaskAzureManagedWorkerBuilder { workItemFilters: this._workItemFilters, }); + for (const middleware of this._orchestrationMiddleware) { + worker.useOrchestrationMiddleware(middleware); + } + + for (const middleware of this._activityMiddleware) { + worker.useActivityMiddleware(middleware); + } + // Register all orchestrators for (const { name, fn } of this._orchestrators) { if (name) { diff --git a/packages/durabletask-js-azuremanaged/test/unit/worker-builder.spec.ts b/packages/durabletask-js-azuremanaged/test/unit/worker-builder.spec.ts index d0f97cb4..cf9b196b 100644 --- a/packages/durabletask-js-azuremanaged/test/unit/worker-builder.spec.ts +++ b/packages/durabletask-js-azuremanaged/test/unit/worker-builder.spec.ts @@ -2,7 +2,13 @@ // Licensed under the MIT License. import { DurableTaskAzureManagedWorkerBuilder, createAzureManagedWorkerBuilder } from "../../src/worker-builder"; -import { TaskEntity, ITaskEntity, TaskEntityOperation } from "@microsoft/durabletask-js"; +import { + ActivityMiddleware, + OrchestrationMiddleware, + TaskEntity, + ITaskEntity, + TaskEntityOperation, +} from "@microsoft/durabletask-js"; // Simple test entity for registration testing class CounterEntity extends TaskEntity { @@ -79,6 +85,22 @@ describe("DurableTaskAzureManagedWorkerBuilder", () => { expect(worker).toBeDefined(); }); + + describe("middleware", () => { + it("supports orchestration and activity middleware registration through the builder", () => { + const orchestrationMiddleware: OrchestrationMiddleware = async (context, next) => next(context); + const activityMiddleware: ActivityMiddleware = async (context, next) => next(context); + const builder = new DurableTaskAzureManagedWorkerBuilder(); + + const result = builder + .endpoint(ENDPOINT, TASKHUB, null) + .useOrchestrationMiddleware(orchestrationMiddleware) + .useActivityMiddleware(activityMiddleware); + + expect(result).toBe(builder); + expect(result.build()).toBeDefined(); + }); + }); }); describe("createAzureManagedWorkerBuilder", () => { diff --git a/packages/durabletask-js/src/index.ts b/packages/durabletask-js/src/index.ts index 1bb5cd3d..bea5bdd2 100644 --- a/packages/durabletask-js/src/index.ts +++ b/packages/durabletask-js/src/index.ts @@ -5,6 +5,17 @@ export { TaskHubGrpcClient, TaskHubGrpcClientOptions, MetadataGenerator } from "./client/client"; export { TaskHubGrpcWorker, TaskHubGrpcWorkerOptions } from "./worker/task-hub-grpc-worker"; export { VersioningOptions, VersionMatchStrategy, VersionFailureStrategy } from "./worker/versioning-options"; +export { + ActivityMiddleware, + ActivityMiddlewareContext, + ActivityMiddlewareNext, + MiddlewareFeature, + MiddlewareFeatures, + OrchestrationMiddleware, + OrchestrationMiddlewareContext, + OrchestrationMiddlewareNext, + createMiddlewareFeature, +} from "./worker/middleware"; export { WorkItemFilters, OrchestrationWorkItemFilter, diff --git a/packages/durabletask-js/src/testing/in-memory-backend.ts b/packages/durabletask-js/src/testing/in-memory-backend.ts index 1275aace..59635cae 100644 --- a/packages/durabletask-js/src/testing/in-memory-backend.ts +++ b/packages/durabletask-js/src/testing/in-memory-backend.ts @@ -6,6 +6,7 @@ import * as pbh from "../utils/pb-helper.util"; import { OrchestrationStatus as ClientOrchestrationStatus } from "../orchestration/enum/orchestration-status.enum"; import { ParentOrchestrationInstance } from "../types/parent-orchestration-instance.type"; import { randomUUID } from "crypto"; +import { mapToRecord } from "../utils/tags.util"; /** Mints a fresh per-execution ID (DTFx `Guid.ToString("N")` idiom: 32 hex chars, no dashes). */ function newExecutionId(): string { @@ -39,6 +40,8 @@ export interface ActivityWorkItem { name: string; taskId: number; input?: string; + version?: string; + tags?: Record; completionToken: number; } @@ -53,12 +56,12 @@ interface StateWaiter { /** * In-memory backend for durable orchestrations suitable for testing. - * + * * This backend stores all orchestration state in memory and processes * work items synchronously within the same process. It is designed for * unit testing and integration testing scenarios where a sidecar process * or external storage is not desired. - * + * * Thread-safety: All state mutations are performed synchronously via * the event loop. The backend uses a simple work queue pattern to ensure * that orchestration and activity processing happens in a predictable order. @@ -301,7 +304,7 @@ export class InMemoryOrchestrationBackend { const instanceId = this.orchestrationQueue.shift()!; this.orchestrationQueueSet.delete(instanceId); const instance = this.instances.get(instanceId); - + if (instance && instance.pendingEvents.length > 0) { return instance; } @@ -375,9 +378,7 @@ export class InMemoryOrchestrationBackend { // Continue-as-new resets status to PENDING and rewind resets it to RUNNING, so neither is // terminal here and neither gets a bookend. if (this.isTerminalStatus(instance.status)) { - instance.history.push( - pbh.newExecutionCompletedEvent(instance.status, instance.output, instance.failureDetails), - ); + instance.history.push(pbh.newExecutionCompletedEvent(instance.status, instance.output, instance.failureDetails)); } // Update completion token for next execution @@ -390,12 +391,7 @@ export class InMemoryOrchestrationBackend { /** * Completes an activity execution. */ - completeActivity( - instanceId: string, - taskId: number, - result?: string, - error?: Error, - ): void { + completeActivity(instanceId: string, taskId: number, result?: string, error?: Error): void { const instance = this.instances.get(instanceId); if (!instance) { return; // Instance may have been purged @@ -623,7 +619,13 @@ export class InMemoryOrchestrationBackend { // because it sets currentUtcDateTime, and ExecutionStarted must come before // carryover events because it initializes the orchestrator generator. const orchestratorStarted = pbh.newOrchestratorStartedEvent(new Date()); - const executionStarted = pbh.newExecutionStartedEvent(instance.name, instance.instanceId, newInput, undefined, instance.executionId); + const executionStarted = pbh.newExecutionStartedEvent( + instance.name, + instance.instanceId, + newInput, + undefined, + instance.executionId, + ); instance.pendingEvents = [orchestratorStarted, executionStarted, ...carryoverEvents]; this.enqueueOrchestration(instance.instanceId); @@ -635,6 +637,8 @@ export class InMemoryOrchestrationBackend { const taskId = action.getId(); const taskName = scheduleTask.getName(); const input = scheduleTask.getInput()?.getValue(); + const version = scheduleTask.getVersion()?.getValue() || undefined; + const tags = mapToRecord(scheduleTask.getTagsMap()); // Add TaskScheduled event to history const event = pbh.newTaskScheduledEvent(taskId, taskName, input); @@ -651,6 +655,8 @@ export class InMemoryOrchestrationBackend { name: taskName, taskId, input, + version, + tags, completionToken: instance.completionToken, }); } diff --git a/packages/durabletask-js/src/testing/test-worker.ts b/packages/durabletask-js/src/testing/test-worker.ts index ed389b8f..82beb052 100644 --- a/packages/durabletask-js/src/testing/test-worker.ts +++ b/packages/durabletask-js/src/testing/test-worker.ts @@ -11,10 +11,11 @@ import { TOutput } from "../types/output.type"; import { InMemoryOrchestrationBackend, OrchestrationInstance, ActivityWorkItem } from "./in-memory-backend"; import * as pb from "../proto/orchestrator_service_pb"; import * as pbh from "../utils/pb-helper.util"; +import { ActivityMiddleware, OrchestrationMiddleware } from "../worker/middleware"; /** * Worker that processes orchestrations and activities from the in-memory backend. - * + * * This worker runs in the same process as the test and processes work items * synchronously in the Node.js event loop, avoiding the need for a separate * sidecar process. @@ -25,6 +26,8 @@ export class TestOrchestrationWorker { private isRunning: boolean = false; private processingPromise: Promise | null = null; private stopRequested: boolean = false; + private readonly orchestrationMiddleware: OrchestrationMiddleware[] = []; + private readonly activityMiddleware: ActivityMiddleware[] = []; constructor(backend: InMemoryOrchestrationBackend) { this.registry = new Registry(); @@ -73,6 +76,28 @@ export class TestOrchestrationWorker { return name; } + /** + * Adds orchestration middleware to this worker. + */ + useOrchestrationMiddleware(middleware: OrchestrationMiddleware): this { + if (this.isRunning) { + throw new Error("Cannot add orchestration middleware while worker is running."); + } + this.orchestrationMiddleware.push(middleware); + return this; + } + + /** + * Adds activity middleware to this worker. + */ + useActivityMiddleware(middleware: ActivityMiddleware): this { + if (this.isRunning) { + throw new Error("Cannot add activity middleware while worker is running."); + } + this.activityMiddleware.push(middleware); + return this; + } + /** * Starts the worker processing loop. */ @@ -141,7 +166,7 @@ export class TestOrchestrationWorker { const completionToken = instance.completionToken; try { - const executor = new OrchestrationExecutor(this.registry); + const executor = new OrchestrationExecutor(this.registry, undefined, this.orchestrationMiddleware); const result = await executor.execute(instanceId, instance.history, instance.pendingEvents, instance.executionId); this.backend.completeOrchestration(instanceId, completionToken, result.actions, result.customStatus); @@ -163,11 +188,11 @@ export class TestOrchestrationWorker { * Processes a single activity work item. */ private async processActivity(workItem: ActivityWorkItem): Promise { - const { instanceId, name, taskId, input } = workItem; + const { instanceId, name, taskId, input, version, tags } = workItem; try { - const executor = new ActivityExecutor(this.registry); - const result = await executor.execute(instanceId, name, taskId, input); + const executor = new ActivityExecutor(this.registry, undefined, this.activityMiddleware); + const result = await executor.execute(instanceId, name, taskId, input, { version, tags }); this.backend.completeActivity(instanceId, taskId, result); } catch (error: unknown) { const err = error instanceof Error ? error : new Error(String(error)); diff --git a/packages/durabletask-js/src/worker/activity-executor.ts b/packages/durabletask-js/src/worker/activity-executor.ts index f4ad151a..cbd253dd 100644 --- a/packages/durabletask-js/src/worker/activity-executor.ts +++ b/packages/durabletask-js/src/worker/activity-executor.ts @@ -7,14 +7,23 @@ import { Logger, ConsoleLogger } from "../types/logger.type"; import { ActivityNotRegisteredError } from "./exception/activity-not-registered-error"; import { Registry } from "./registry"; import * as WorkerLogs from "./logs"; +import { + ActivityExecutionOptions, + ActivityMiddleware, + DefaultActivityMiddlewareContext, + MiddlewareFeatures, + runActivityMiddleware, +} from "./middleware"; export class ActivityExecutor { private _registry: Registry; private _logger: Logger; + private _middleware: readonly ActivityMiddleware[]; - constructor(registry: Registry, logger?: Logger) { + constructor(registry: Registry, logger?: Logger, middleware: readonly ActivityMiddleware[] = []) { this._registry = registry; this._logger = logger ?? new ConsoleLogger(); + this._middleware = [...middleware]; } public async execute( @@ -22,6 +31,7 @@ export class ActivityExecutor { name: string, taskId: number, encodedInput?: string, + options: ActivityExecutionOptions = {}, ): Promise { const fn = this._registry.getActivity(name); @@ -32,23 +42,32 @@ export class ActivityExecutor { // Log activity start (EventId 603) WorkerLogs.activityStarted(this._logger, orchestrationId, name); - const ctx = new ActivityContext(orchestrationId, taskId); - try { // Deserialize the input inside the try-catch so that malformed JSON // is reported through the same activityFailed log path (EventId 605) // as any other activity execution error. const activityInput = encodedInput ? JSON.parse(encodedInput) : undefined; + const ctx = new ActivityContext(orchestrationId, taskId); + const middlewareContext = new DefaultActivityMiddlewareContext( + name, + orchestrationId, + taskId, + options.version, + options.tags, + activityInput, + encodedInput || undefined, + ctx, + options.features ?? new MiddlewareFeatures(), + ); - // Execute the activity function - let activityOutput = fn(ctx, activityInput); - - if (isPromise(activityOutput)) { - activityOutput = await activityOutput; - } + await runActivityMiddleware(middlewareContext, this._middleware, async () => { + const activityOutput = fn(ctx, activityInput); + return isPromise(activityOutput) ? await activityOutput : activityOutput; + }); // Return the output - const encodedOutput = activityOutput !== undefined ? JSON.stringify(activityOutput) : undefined; + const encodedOutput = + middlewareContext.result !== undefined ? JSON.stringify(middlewareContext.result) : undefined; // Log activity completion (EventId 604) WorkerLogs.activityCompleted(this._logger, orchestrationId, name); diff --git a/packages/durabletask-js/src/worker/middleware.ts b/packages/durabletask-js/src/worker/middleware.ts new file mode 100644 index 00000000..e9fa55ae --- /dev/null +++ b/packages/durabletask-js/src/worker/middleware.ts @@ -0,0 +1,366 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { ActivityContext } from "../task/context/activity-context"; +import { OrchestrationContext } from "../task/context/orchestration-context"; +import { ParentOrchestrationInstance } from "../types/parent-orchestration-instance.type"; + +declare const middlewareFeatureType: unique symbol; + +/** + * A typed symbol used to store a host-specific object in {@link MiddlewareFeatures}. + */ +export type MiddlewareFeature = symbol & { + readonly [middlewareFeatureType]?: T; +}; + +/** + * Creates a typed key for a host-specific middleware feature. + */ +export function createMiddlewareFeature(description?: string): MiddlewareFeature { + return Symbol(description) as MiddlewareFeature; +} + +/** + * Per-work-item host objects available to durable middleware. + * + * Feature values are process-local and are never serialized into durable history. + */ +export class MiddlewareFeatures { + readonly #features = new Map(); + + get(key: MiddlewareFeature): T | undefined { + return this.#features.get(key) as T | undefined; + } + + set(key: MiddlewareFeature, value: T): this { + this.#features.set(key, value); + return this; + } + + has(key: symbol): boolean { + return this.#features.has(key); + } + + delete(key: symbol): boolean { + return this.#features.delete(key); + } +} + +export interface OrchestrationMiddlewareContext { + readonly name: string; + readonly instanceId: string; + readonly version?: string; + readonly parent?: ParentOrchestrationInstance; + readonly tags?: Readonly>; + readonly input: TInput; + readonly rawInput?: string; + readonly isReplaying: boolean; + readonly orchestrationContext: OrchestrationContext; + readonly features: MiddlewareFeatures; + readonly result?: TResult; + readonly failure?: Error; +} + +export type OrchestrationMiddlewareNext = (context: OrchestrationMiddlewareContext) => Promise; + +export type OrchestrationMiddleware = ( + context: OrchestrationMiddlewareContext, + next: OrchestrationMiddlewareNext, +) => Promise; + +export interface ActivityMiddlewareContext { + readonly name: string; + readonly instanceId: string; + readonly taskId: number; + readonly version?: string; + readonly tags?: Readonly>; + readonly input: TInput; + readonly rawInput?: string; + readonly activityContext: ActivityContext; + readonly features: MiddlewareFeatures; + readonly result?: TResult; + readonly failure?: Error; + setResult(result: TResult): void; +} + +export type ActivityMiddlewareNext = (context: ActivityMiddlewareContext) => Promise; + +export type ActivityMiddleware = (context: ActivityMiddlewareContext, next: ActivityMiddlewareNext) => Promise; + +export interface ActivityExecutionOptions { + version?: string; + tags?: Readonly>; + features?: MiddlewareFeatures; +} + +export class DefaultOrchestrationMiddlewareContext implements OrchestrationMiddlewareContext { + private _result?: unknown; + private _failure?: Error; + + constructor( + readonly name: string, + readonly instanceId: string, + readonly version: string | undefined, + readonly parent: ParentOrchestrationInstance | undefined, + tags: Readonly> | undefined, + readonly input: unknown, + readonly rawInput: string | undefined, + readonly isReplaying: boolean, + readonly orchestrationContext: OrchestrationContext, + readonly features: MiddlewareFeatures, + ) { + this.tags = copyTags(tags); + } + + readonly tags?: Readonly>; + + get result(): unknown { + return this._result; + } + + get failure(): Error | undefined { + return this._failure; + } + + setExecutionResult(result: unknown, failure?: Error): void { + this._result = result; + this._failure = failure; + } +} + +export class DefaultActivityMiddlewareContext implements ActivityMiddlewareContext { + private _result?: unknown; + private _failure?: Error; + private _resultRevision = 0; + + constructor( + readonly name: string, + readonly instanceId: string, + readonly taskId: number, + readonly version: string | undefined, + tags: Readonly> | undefined, + readonly input: unknown, + readonly rawInput: string | undefined, + readonly activityContext: ActivityContext, + readonly features: MiddlewareFeatures, + ) { + this.tags = copyTags(tags); + } + + readonly tags?: Readonly>; + + get result(): unknown { + return this._result; + } + + get failure(): Error | undefined { + return this._failure; + } + + get resultRevision(): number { + return this._resultRevision; + } + + get hasResult(): boolean { + return this._resultRevision > 0; + } + + setResult(result: unknown): void { + this._result = result; + this._failure = undefined; + this._resultRevision++; + } + + setFailure(error: Error): void { + this._failure = error; + } +} + +export async function runOrchestrationMiddleware( + context: DefaultOrchestrationMiddlewareContext, + middleware: readonly OrchestrationMiddleware[], + body: () => Promise, + suspended?: Promise, +): Promise<"completed" | "suspended"> { + const validation = new MiddlewareValidation(); + + const invoke = async (index: number): Promise => { + if (index === middleware.length) { + await body(); + return; + } + + let nextCalls = 0; + let nextStarted = false; + let nextSettled = false; + let nextPromise: Promise | undefined; + await middleware[index](context, (nextContext) => { + if (nextContext !== context || nextCalls++ !== 0) { + const error = new Error("Orchestration middleware must call next exactly once."); + validation.record(error); + return createLazyPromise(() => Promise.reject(error)); + } + + nextPromise = createLazyPromise(async () => { + nextStarted = true; + try { + await invoke(index + 1); + } finally { + nextSettled = true; + } + }); + return nextPromise; + }); + + if (nextCalls !== 1) { + validation.record(new Error("Orchestration middleware must call next exactly once.")); + } else if (!nextStarted || !nextSettled) { + validation.record(new Error("Orchestration middleware must not return before next(context) completes.")); + if (nextStarted && nextPromise) { + if (suspended) { + await Promise.race([nextPromise.catch(() => {}), suspended]); + } else { + await nextPromise.catch(() => {}); + } + } + } + + validation.throwIfFailed(); + }; + + try { + const invocation = invoke(0).then(() => "completed" as const); + const outcome = suspended + ? await Promise.race([invocation, suspended.then(() => "suspended" as const)]) + : await invocation; + if (outcome === "suspended") { + await Promise.resolve(); + } + validation.throwIfFailed(); + return outcome; + } catch (error: unknown) { + validation.throwIfFailed(); + throw error; + } +} + +export async function runActivityMiddleware( + context: DefaultActivityMiddlewareContext, + middleware: readonly ActivityMiddleware[], + body: () => Promise, +): Promise { + const validation = new MiddlewareValidation(); + + const invoke = async (index: number): Promise => { + if (index === middleware.length) { + if (!context.hasResult) { + try { + context.setResult(await body()); + } catch (error: unknown) { + const failure = error instanceof Error ? error : new Error(String(error)); + context.setFailure(failure); + throw failure; + } + } + return; + } + + let nextCalls = 0; + let nextStarted = false; + let nextSettled = false; + let nextPromise: Promise | undefined; + const resultRevision = context.resultRevision; + await middleware[index](context, (nextContext) => { + if (nextContext !== context || nextCalls++ !== 0) { + const error = new Error("Activity middleware must call next at most once."); + validation.record(error); + return createLazyPromise(() => Promise.reject(error)); + } + if (context.hasResult) { + const error = new Error("Activity middleware cannot call next after setResult."); + validation.record(error); + return createLazyPromise(() => Promise.reject(error)); + } + + nextPromise = createLazyPromise(async () => { + nextStarted = true; + try { + await invoke(index + 1); + } finally { + nextSettled = true; + } + }); + return nextPromise; + }); + + if (nextCalls === 0 && context.resultRevision === resultRevision) { + validation.record(new Error("Activity middleware must call next exactly once or call setResult.")); + } else if (nextPromise && (!nextStarted || !nextSettled)) { + validation.record(new Error("Activity middleware must not return before next(context) completes.")); + if (nextStarted) { + await nextPromise; + } + } + + validation.throwIfFailed(); + }; + + try { + await invoke(0); + validation.throwIfFailed(); + if (context.failure) { + throw context.failure; + } + } catch (error: unknown) { + const failure = validation.failure ?? (error instanceof Error ? error : new Error(String(error))); + context.setFailure(failure); + throw failure; + } +} + +function copyTags(tags: Readonly> | undefined): Readonly> | undefined { + return tags ? Object.freeze({ ...tags }) : undefined; +} + +class MiddlewareValidation { + private _failure?: Error; + + get failure(): Error | undefined { + return this._failure; + } + + record(error: Error): void { + this._failure ??= error; + } + + throwIfFailed(): void { + if (this._failure) { + throw this._failure; + } + } +} + +function createLazyPromise(start: () => Promise): Promise { + let promise: Promise | undefined; + const getPromise = (): Promise => { + promise ??= start(); + return promise; + }; + const observe = (result: Promise): Promise => { + void result.catch(() => {}); + return { + then: (onFulfilled, onRejected) => observe(result.then(onFulfilled, onRejected)), + catch: (onRejected) => observe(result.catch(onRejected)), + finally: (onFinally) => observe(result.finally(onFinally)), + [Symbol.toStringTag]: "Promise", + } as Promise; + }; + + return { + then: (onFulfilled, onRejected) => observe(getPromise().then(onFulfilled, onRejected)), + catch: (onRejected) => observe(getPromise().catch(onRejected)), + finally: (onFinally) => observe(getPromise().finally(onFinally)), + [Symbol.toStringTag]: "Promise", + } as Promise; +} diff --git a/packages/durabletask-js/src/worker/orchestration-executor.ts b/packages/durabletask-js/src/worker/orchestration-executor.ts index 81b4b836..6189b5af 100644 --- a/packages/durabletask-js/src/worker/orchestration-executor.ts +++ b/packages/durabletask-js/src/worker/orchestration-executor.ts @@ -31,6 +31,13 @@ import { EntityOperationFailedException, createTaskFailureDetails, } from "../entities/entity-operation-failed-exception"; +import { mapToRecord } from "../utils/tags.util"; +import { + DefaultOrchestrationMiddlewareContext, + MiddlewareFeatures, + OrchestrationMiddleware, + runOrchestrationMiddleware, +} from "./middleware"; /** * Result of orchestration execution containing actions and optional custom status. @@ -47,14 +54,16 @@ export class OrchestrationExecutor { private _suspendedEvents: pb.HistoryEvent[]; private _logger: Logger; private _orchestratorName: string; + private _middleware: readonly OrchestrationMiddleware[]; - constructor(registry: Registry, logger?: Logger) { + constructor(registry: Registry, logger?: Logger, middleware: readonly OrchestrationMiddleware[] = []) { this._registry = registry; this._generator = undefined; this._isSuspended = false; this._suspendedEvents = []; this._logger = logger ?? new ConsoleLogger(); this._orchestratorName = "(unknown)"; + this._middleware = [...middleware]; } async execute( @@ -62,6 +71,7 @@ export class OrchestrationExecutor { oldEvents: pb.HistoryEvent[], newEvents: pb.HistoryEvent[], executionId?: string, + features: MiddlewareFeatures = new MiddlewareFeatures(), ): Promise { if (!newEvents?.length) { throw new OrchestrationStateError("The new history event list must have at least one event in it"); @@ -92,22 +102,80 @@ export class OrchestrationExecutor { ctx._executionId = executionId; } - try { - // Rebuild the local state by replaying the history events into the orchestrator function - WorkerLogs.orchestrationRebuilding(this._logger, instanceId, oldEvents.length); - ctx._isReplaying = true; - - for (const oldEvent of oldEvents) { - await this.processEvent(ctx, oldEvent); - } + const allEvents = [...oldEvents, ...newEvents]; + const executionStarted = allEvents.find((event) => event.hasExecutionstarted())?.getExecutionstarted(); + const initialOrchestratorStarted = allEvents.find((event) => event.hasOrchestratorstarted()); + const rawInput = isEmpty(executionStarted?.getInput()) ? undefined : executionStarted?.getInput()?.getValue(); + const parentInstance = executionStarted?.getParentinstance(); + const parentOrchestrationInstance = parentInstance?.getOrchestrationinstance(); + ctx._version = executionStarted?.getVersion()?.getValue() ?? ""; + ctx._parent = parentInstance + ? { + name: parentInstance.getName()?.getValue() ?? "", + instanceId: parentOrchestrationInstance?.getInstanceid() ?? "", + taskScheduledId: parentInstance.getTaskscheduledid(), + } + : undefined; + ctx._isReplaying = oldEvents.length > 0; + ctx._currentUtcDatetime = initialOrchestratorStarted?.getTimestamp()?.toDate() ?? ctx._currentUtcDatetime; + this._orchestratorName = executionStarted?.getName() ?? "(unknown)"; - // Get new actions by executing newly received events into the orchestrator function - const summary = getNewEventSummary(newEvents); - WorkerLogs.orchestrationProcessing(this._logger, instanceId, newEvents.length, summary); - ctx._isReplaying = false; + try { + const middlewareContext = new DefaultOrchestrationMiddlewareContext( + this._orchestratorName, + instanceId, + ctx._version || undefined, + ctx._parent, + mapToRecord(executionStarted?.getTagsMap()), + parseJsonField(executionStarted?.getInput()), + rawInput, + oldEvents.length > 0, + ctx, + features, + ); - for (const newEvent of newEvents) { - await this.processEvent(ctx, newEvent); + try { + let signalSuspended!: () => void; + const suspended = new Promise((resolve) => { + signalSuspended = resolve; + }); + await runOrchestrationMiddleware( + middlewareContext, + this._middleware, + async () => { + try { + // Rebuild the local state by replaying the history events into the orchestrator function + WorkerLogs.orchestrationRebuilding(this._logger, instanceId, oldEvents.length); + ctx._isReplaying = true; + + for (const oldEvent of oldEvents) { + await this.processEvent(ctx, oldEvent); + } + + // Get new actions by executing newly received events into the orchestrator function + const summary = getNewEventSummary(newEvents); + WorkerLogs.orchestrationProcessing(this._logger, instanceId, newEvents.length, summary); + ctx._isReplaying = false; + + for (const newEvent of newEvents) { + await this.processEvent(ctx, newEvent); + } + } catch (e: unknown) { + ctx.setFailed(e instanceof Error ? e : new Error(String(e))); + } + + middlewareContext.setExecutionResult(ctx._result, ctx._failure); + if (!ctx._isComplete) { + signalSuspended(); + await new Promise(() => {}); + } + }, + suspended, + ); + } catch (e: unknown) { + const error = e instanceof Error ? e : new Error(String(e)); + ctx.setFailed(error); + middlewareContext.setExecutionResult(undefined, error); } } catch (e: unknown) { ctx.setFailed(e instanceof Error ? e : new Error(String(e))); @@ -250,9 +318,7 @@ export class OrchestrationExecutor { private async handleExecutionStarted(ctx: RuntimeOrchestrationContext, event: pb.HistoryEvent): Promise { // TODO: Check if we already started the orchestration const executionStartedEvent = event.getExecutionstarted(); - const fn = this._registry.getOrchestrator( - executionStartedEvent ? executionStartedEvent.getName() : undefined, - ); + const fn = this._registry.getOrchestrator(executionStartedEvent ? executionStartedEvent.getName() : undefined); if (!fn) { throw new OrchestratorNotRegisteredError(executionStartedEvent?.getName()); @@ -419,9 +485,7 @@ export class OrchestrationExecutor { } else if (!isCreateSubOrchestrationAction) { const expectedMethodName = getName(ctx.callSubOrchestrator); throw getWrongActionTypeError(taskId, expectedMethodName, action); - } else if ( - action.getCreatesuborchestration()?.getName() != event.getSuborchestrationinstancecreated()?.getName() - ) { + } else if (action.getCreatesuborchestration()?.getName() != event.getSuborchestrationinstancecreated()?.getName()) { throw getWrongActionNameError( taskId, getName(ctx.callSubOrchestrator), @@ -431,7 +495,10 @@ export class OrchestrationExecutor { } } - private async handleSubOrchestrationCompleted(ctx: RuntimeOrchestrationContext, event: pb.HistoryEvent): Promise { + private async handleSubOrchestrationCompleted( + ctx: RuntimeOrchestrationContext, + event: pb.HistoryEvent, + ): Promise { const completedEvent = event.getSuborchestrationinstancecompleted(); const taskId = completedEvent ? completedEvent.getTaskscheduledid() : undefined; const result = completedEvent?.getResult(); @@ -445,7 +512,13 @@ export class OrchestrationExecutor { ? subOrchestrationInstanceFailedEvent.getTaskscheduledid() : undefined; const failureDetails = subOrchestrationInstanceFailedEvent?.getFailuredetails(); - await this.handleFailedTask(ctx, taskId, failureDetails, "subOrchestrationInstanceFailed", "Sub-orchestration task"); + await this.handleFailedTask( + ctx, + taskId, + failureDetails, + "subOrchestrationInstanceFailed", + "Sub-orchestration task", + ); } private async handleEventRaised(ctx: RuntimeOrchestrationContext, event: pb.HistoryEvent): Promise { @@ -751,17 +824,15 @@ export class OrchestrationExecutor { ); } - private async handleEntityOperationCompleted(ctx: RuntimeOrchestrationContext, event: pb.HistoryEvent): Promise { + private async handleEntityOperationCompleted( + ctx: RuntimeOrchestrationContext, + event: pb.HistoryEvent, + ): Promise { const completedEvent = event.getEntityoperationcompleted(); const requestId = completedEvent?.getRequestid(); if (!requestId) { - WorkerLogs.entityEventIgnored( - this._logger, - ctx._instanceId, - "EntityOperationCompletedEvent", - "no requestId", - ); + WorkerLogs.entityEventIgnored(this._logger, ctx._instanceId, "EntityOperationCompletedEvent", "no requestId"); return; } @@ -798,12 +869,7 @@ export class OrchestrationExecutor { const requestId = failedEvent?.getRequestid(); if (!requestId) { - WorkerLogs.entityEventIgnored( - this._logger, - ctx._instanceId, - "EntityOperationFailedEvent", - "no requestId", - ); + WorkerLogs.entityEventIgnored(this._logger, ctx._instanceId, "EntityOperationFailedEvent", "no requestId"); return; } @@ -828,12 +894,10 @@ export class OrchestrationExecutor { // If in a critical section, recover the lock for this entity ctx._entityFeature.recoverLockAfterCall(pendingCall.entityId); - const failureDetails = - createTaskFailureDetails(failedEvent?.getFailuredetails()) ?? - { - errorType: "UnknownError", - errorMessage: `Entity operation '${pendingCall.operationName}' failed with unknown error`, - }; + const failureDetails = createTaskFailureDetails(failedEvent?.getFailuredetails()) ?? { + errorType: "UnknownError", + errorMessage: `Entity operation '${pendingCall.operationName}' failed with unknown error`, + }; const exception = new EntityOperationFailedException( pendingCall.entityId, pendingCall.operationName, @@ -849,12 +913,7 @@ export class OrchestrationExecutor { const criticalSectionId = lockGrantedEvent?.getCriticalsectionid(); if (!criticalSectionId) { - WorkerLogs.entityEventIgnored( - this._logger, - ctx._instanceId, - "EntityLockGrantedEvent", - "no criticalSectionId", - ); + WorkerLogs.entityEventIgnored(this._logger, ctx._instanceId, "EntityLockGrantedEvent", "no criticalSectionId"); return; } @@ -973,15 +1032,11 @@ export class OrchestrationExecutor { // No retry - fail the task delete ctx._pendingTasks[taskId]; - task.fail( - `${taskLabel} #${taskId} failed: ${errorMessage}`, - failureDetails, - ); + task.fail(`${taskLabel} #${taskId} failed: ${errorMessage}`, failureDetails); await ctx.resume(); } - /** * Checks if a failed task supports retry and handles the retry if applicable. * Supports both RetryableTask (policy-based with timer delay) and RetryHandlerTask diff --git a/packages/durabletask-js/src/worker/runtime-orchestration-context.ts b/packages/durabletask-js/src/worker/runtime-orchestration-context.ts index 828deb4e..eccb87ac 100644 --- a/packages/durabletask-js/src/worker/runtime-orchestration-context.ts +++ b/packages/durabletask-js/src/worker/runtime-orchestration-context.ts @@ -21,11 +21,7 @@ import { Task } from "../task/task"; import { StopIterationError } from "./exception/stop-iteration-error"; import { mapToRecord } from "../utils/tags.util"; -import { - OrchestrationEntityFeature, - CriticalSectionInfo, - LockHandle, -} from "../entities/orchestration-entity-feature"; +import { OrchestrationEntityFeature, CriticalSectionInfo, LockHandle } from "../entities/orchestration-entity-feature"; import { EntityInstanceId } from "../entities/entity-instance-id"; import { SignalEntityOptions, CallEntityOptions } from "../entities/signal-entity-options"; @@ -35,6 +31,7 @@ export class RuntimeOrchestrationContext extends OrchestrationContext { _isReplaying: boolean; _isComplete: boolean; _result: any; + _failure?: Error; _pendingActions: Record; _pendingTasks: Record>; _sequenceNumber: number; @@ -59,6 +56,7 @@ export class RuntimeOrchestrationContext extends OrchestrationContext { this._isReplaying = true; this._isComplete = false; this._result = undefined; + this._failure = undefined; this._pendingActions = {}; this._pendingTasks = {}; this._sequenceNumber = 0; @@ -221,6 +219,7 @@ export class RuntimeOrchestrationContext extends OrchestrationContext { // must be preserved and returned alongside the complete action this._result = result; + this._failure = undefined; let resultJson; @@ -246,6 +245,7 @@ export class RuntimeOrchestrationContext extends OrchestrationContext { this._isComplete = true; this._completionStatus = pb.OrchestrationStatus.ORCHESTRATION_STATUS_FAILED; + this._failure = e; // Note: Do NOT clear pending actions here - fire-and-forget actions like sendEvent // must be preserved and returned alongside the complete action @@ -329,15 +329,11 @@ export class RuntimeOrchestrationContext extends OrchestrationContext { } else if (fireAt instanceof Date) { fireAtDate = fireAt; } else { - throw new Error( - `createTimer requires a finite number (seconds) or a valid Date, but received ${String(fireAt)}`, - ); + throw new Error(`createTimer requires a finite number (seconds) or a valid Date, but received ${String(fireAt)}`); } if (Number.isNaN(fireAtDate.getTime())) { - throw new Error( - "createTimer received or produced an invalid Date (NaN timestamp)", - ); + throw new Error("createTimer received or produced an invalid Date (NaN timestamp)"); } const action = ph.newCreateTimerAction(id, fireAtDate); @@ -415,7 +411,14 @@ export class RuntimeOrchestrationContext extends OrchestrationContext { } const encodedInput = input !== undefined ? JSON.stringify(input) : undefined; - const action = ph.newCreateSubOrchestrationAction(id, name, instanceId, encodedInput, options?.tags, options?.version); + const action = ph.newCreateSubOrchestrationAction( + id, + name, + instanceId, + encodedInput, + options?.tags, + options?.version, + ); this._pendingActions[action.getId()] = action; const task = this.createRetryTaskOrDefault(action, id, options, "subOrchestration"); @@ -485,10 +488,9 @@ export class RuntimeOrchestrationContext extends OrchestrationContext { try { this._customStatus = JSON.stringify(customStatus); } catch (e) { - throw new Error( - `Custom status value is not JSON-serializable: ${e instanceof Error ? e.message : String(e)}`, - { cause: e }, - ); + throw new Error(`Custom status value is not JSON-serializable: ${e instanceof Error ? e.message : String(e)}`, { + cause: e, + }); } } @@ -640,15 +642,10 @@ export class RuntimeOrchestrationContext extends OrchestrationContext { action: pb.OrchestratorAction, id: number, options: TaskOptions | SubOrchestrationOptions | undefined, - taskType: RetryTaskType + taskType: RetryTaskType, ): CompletableTask { if (options?.retry && isRetryPolicy(options.retry)) { - const retryableTask = new RetryableTask( - options.retry, - action, - this._currentUtcDatetime, - taskType, - ); + const retryableTask = new RetryableTask(options.retry, action, this._currentUtcDatetime, taskType); this._pendingTasks[id] = retryableTask; return retryableTask; } @@ -657,13 +654,7 @@ export class RuntimeOrchestrationContext extends OrchestrationContext { // Normalize to AsyncRetryHandler — wraps sync handlers via Promise.resolve, // and is a no-op for handlers that already return a Promise. const handler = toAsyncRetryHandler(options.retry); - const retryHandlerTask = new RetryHandlerTask( - handler, - this, - action, - this._currentUtcDatetime, - taskType, - ); + const retryHandlerTask = new RetryHandlerTask(handler, this, action, this._currentUtcDatetime, taskType); this._pendingTasks[id] = retryHandlerTask; return retryHandlerTask; } @@ -744,10 +735,7 @@ class RuntimeOrchestrationEntityFeature implements OrchestrationEntityFeature { * Tracks pending lock acquisitions by criticalSectionId. * Used to correlate EntityLockGranted events with the original lock request. */ - readonly pendingLockRequests: Map< - string, - { task: CompletableTask; lockSet: EntityInstanceId[] } - >; + readonly pendingLockRequests: Map; lockSet: EntityInstanceId[] }>; /** * Current critical section state. Null if not in a critical section. @@ -810,25 +798,19 @@ class RuntimeOrchestrationEntityFeature implements OrchestrationEntityFeature { if (this.criticalSection) { // Check if lock acquisition is still pending if (this.lockAcquisitionPending) { - throw new Error( - "Must await the completion of the lock request prior to calling any entity.", - ); + throw new Error("Must await the completion of the lock request prior to calling any entity."); } const entityIdStr = id.toString(); if (!this.criticalSection.availableEntities.has(entityIdStr)) { // Check if this entity is even in the lock set - const isLocked = this.criticalSection.lockedEntities.some( - (e) => e.toString() === entityIdStr, - ); + const isLocked = this.criticalSection.lockedEntities.some((e) => e.toString() === entityIdStr); if (isLocked) { throw new Error( "Must not call an entity from a critical section while a prior call to the same entity is still pending.", ); } else { - throw new Error( - "Must not call an entity from a critical section if it is not one of the locked entities.", - ); + throw new Error("Must not call an entity from a critical section if it is not one of the locked entities."); } } // Mark entity as unavailable until call completes @@ -885,18 +867,11 @@ class RuntimeOrchestrationEntityFeature implements OrchestrationEntityFeature { * This creates a SendEntityMessageAction with an EntityOperationSignaledEvent. * The orchestration does not wait for the entity to process the operation. */ - signalEntity( - id: EntityInstanceId, - operationName: string, - input?: unknown, - options?: SignalEntityOptions, - ): void { + signalEntity(id: EntityInstanceId, operationName: string, input?: unknown, options?: SignalEntityOptions): void { // Validate: cannot signal a locked entity from within a critical section if (this.criticalSection) { const entityIdStr = id.toString(); - const isLocked = this.criticalSection.lockedEntities.some( - (e) => e.toString() === entityIdStr, - ); + const isLocked = this.criticalSection.lockedEntities.some((e) => e.toString() === entityIdStr); if (isLocked) { throw new Error("Must not signal a locked entity from a critical section."); } @@ -952,10 +927,7 @@ class RuntimeOrchestrationEntityFeature implements OrchestrationEntityFeature { const uniqueEntities: EntityInstanceId[] = []; for (const entity of sortedEntities) { const entityStr = entity.toString(); - if ( - uniqueEntities.length === 0 || - uniqueEntities[uniqueEntities.length - 1].toString() !== entityStr - ) { + if (uniqueEntities.length === 0 || uniqueEntities[uniqueEntities.length - 1].toString() !== entityStr) { uniqueEntities.push(entity); } } @@ -965,12 +937,7 @@ class RuntimeOrchestrationEntityFeature implements OrchestrationEntityFeature { const lockSet = uniqueEntities.map((e) => e.toString()); const parentInstanceId = this.context.instanceId; - const action = ph.newSendEntityMessageLockAction( - actionId, - criticalSectionId, - lockSet, - parentInstanceId, - ); + const action = ph.newSendEntityMessageLockAction(actionId, criticalSectionId, lockSet, parentInstanceId); this.context._pendingActions[action.getId()] = action; @@ -1004,9 +971,7 @@ class RuntimeOrchestrationEntityFeature implements OrchestrationEntityFeature { // Now that lock is granted, populate availableEntities and clear pending flag if (this.criticalSection) { - this.criticalSection.availableEntities = new Set( - pendingRequest.lockSet.map((e) => e.toString()), - ); + this.criticalSection.availableEntities = new Set(pendingRequest.lockSet.map((e) => e.toString())); } this.lockAcquisitionPending = false; @@ -1095,4 +1060,4 @@ class EntityLockReleaser implements LockHandle { this.released = true; this.entityFeature.exitCriticalSection(this.criticalSectionId); } -} \ No newline at end of file +} 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 56bad094..98f6761f 100644 --- a/packages/durabletask-js/src/worker/task-hub-grpc-worker.ts +++ b/packages/durabletask-js/src/worker/task-hub-grpc-worker.ts @@ -37,6 +37,8 @@ import { setSpanOk, endSpan, } from "../tracing"; +import { ActivityMiddleware, MiddlewareFeatures, OrchestrationMiddleware } from "./middleware"; +import { mapToRecord } from "../utils/tags.util"; /** Default timeout in milliseconds for graceful shutdown. */ const DEFAULT_SHUTDOWN_TIMEOUT_MS = 30000; @@ -88,6 +90,8 @@ export class TaskHubGrpcWorker { private _backoff: ExponentialBackoff; private _versioning?: VersioningOptions; private _workItemFilters?: WorkItemFilters | "auto"; + private _orchestrationMiddleware: OrchestrationMiddleware[]; + private _activityMiddleware: ActivityMiddleware[]; /** * Creates a new TaskHubGrpcWorker instance. @@ -179,6 +183,8 @@ export class TaskHubGrpcWorker { }); this._versioning = resolvedVersioning; this._workItemFilters = resolvedWorkItemFilters; + this._orchestrationMiddleware = []; + this._activityMiddleware = []; } /** @@ -277,6 +283,34 @@ export class TaskHubGrpcWorker { return name; } + /** + * Adds orchestration middleware to this worker. + * + * The first registered middleware is the outermost middleware. + */ + useOrchestrationMiddleware(middleware: OrchestrationMiddleware): this { + if (this._isRunning) { + throw new Error("Cannot add orchestration middleware while worker is running."); + } + + this._orchestrationMiddleware.push(middleware); + return this; + } + + /** + * Adds activity middleware to this worker. + * + * The first registered middleware is the outermost middleware. + */ + useActivityMiddleware(middleware: ActivityMiddleware): this { + if (this._isRunning) { + throw new Error("Cannot add activity middleware while worker is running."); + } + + this._activityMiddleware.push(middleware); + return this; + } + /** * Registers an entity with the worker. * @@ -327,10 +361,13 @@ export class TaskHubGrpcWorker { * 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 { + async processOrchestratorRequest( + request: Uint8Array, + features: MiddlewareFeatures = new MiddlewareFeatures(), + ): Promise { const req = pb.OrchestratorRequest.deserializeBinary(request); const stub = new CapturingSidecarStub(); - await this._executeOrchestratorInternal(req, "", stub as unknown as stubs.TaskHubSidecarServiceClient); + await this._executeOrchestratorInternal(req, "", stub as unknown as stubs.TaskHubSidecarServiceClient, features); if (!stub.orchestratorResponse) { if (stub.abandoned) { @@ -353,6 +390,29 @@ export class TaskHubGrpcWorker { return stub.orchestratorResponse.serializeBinary(); } + /** + * Processes a single serialized TaskHubSidecarService ActivityRequest and returns + * the serialized ActivityResponse. + * + * @remarks + * Host integrations can pass per-invocation features without serializing them into + * the durable task request or history. + */ + async processActivityRequest( + request: Uint8Array, + features: MiddlewareFeatures = new MiddlewareFeatures(), + ): Promise { + const req = pb.ActivityRequest.deserializeBinary(request); + const stub = new CapturingSidecarStub(); + await this._executeActivityInternal(req, "", stub as unknown as stubs.TaskHubSidecarServiceClient, features); + + if (!stub.activityResponse) { + throw new Error("Activity execution did not produce a response."); + } + + return stub.activityResponse.serializeBinary(); + } + /** * Processes a single serialized TaskHubSidecarService EntityBatchRequest and * returns the serialized EntityBatchResult. @@ -706,6 +766,7 @@ export class TaskHubGrpcWorker { req: pb.OrchestratorRequest, completionToken: string, stub: stubs.TaskHubSidecarServiceClient, + features: MiddlewareFeatures = new MiddlewareFeatures(), ): Promise { const instanceId = req.getInstanceid(); @@ -809,12 +870,13 @@ export class TaskHubGrpcWorker { let res; try { - const executor = new OrchestrationExecutor(this._registry, this._logger); + const executor = new OrchestrationExecutor(this._registry, this._logger, this._orchestrationMiddleware); const result = await executor.execute( req.getInstanceid(), req.getPasteventsList(), req.getNeweventsList(), req.getExecutionid()?.getValue(), + features, ); // Process actions to inject trace context into scheduled tasks, sub-orchestrations, etc. @@ -902,6 +964,7 @@ export class TaskHubGrpcWorker { req: pb.ActivityRequest, completionToken: string, stub: stubs.TaskHubSidecarServiceClient, + features: MiddlewareFeatures = new MiddlewareFeatures(), ): Promise { const instanceId = req.getOrchestrationinstance()?.getInstanceid(); @@ -915,12 +978,17 @@ export class TaskHubGrpcWorker { const activitySpan = startSpanForTaskExecution(req); try { - const executor = new ActivityExecutor(this._registry, this._logger); + const executor = new ActivityExecutor(this._registry, this._logger, this._activityMiddleware); const result = await executor.execute( instanceId, req.getName(), req.getTaskid(), req.getInput()?.getValue() ?? "", + { + version: req.getVersion()?.getValue() || undefined, + tags: mapToRecord(req.getTagsMap()), + features, + }, ); const s = new StringValue(); @@ -1248,6 +1316,7 @@ export class TaskHubGrpcWorker { */ class CapturingSidecarStub { orchestratorResponse?: pb.OrchestratorResponse; + activityResponse?: pb.ActivityResponse; entityResult?: pb.EntityBatchResult; /** Set when the execution path abandons the work item (e.g. a version mismatch) rather than completing it. */ abandoned = false; @@ -1270,6 +1339,15 @@ class CapturingSidecarStub { callback(null, new Empty()); } + completeActivityTask( + request: pb.ActivityResponse, + _metadata: grpc.Metadata, + callback: (error: grpc.ServiceError | null, response: Empty) => void, + ): void { + this.activityResponse = request; + callback(null, new Empty()); + } + abandonTaskOrchestratorWorkItem( _request: pb.AbandonOrchestrationTaskRequest, _metadata: grpc.Metadata, diff --git a/packages/durabletask-js/test/functions-grpc-support.spec.ts b/packages/durabletask-js/test/functions-grpc-support.spec.ts index c6a55717..c7fe9e04 100644 --- a/packages/durabletask-js/test/functions-grpc-support.spec.ts +++ b/packages/durabletask-js/test/functions-grpc-support.spec.ts @@ -3,15 +3,19 @@ import { OrchestrationContext, + ActivityContext, + MiddlewareFeatures, TaskEntity, TaskHubGrpcWorker, TOrchestrator, VersionFailureStrategy, VersionMatchStrategy, + createMiddlewareFeature, } 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"; +import { StringValue } from "google-protobuf/google/protobuf/wrappers_pb"; const TEST_INSTANCE_ID = "functions-grpc-instance"; @@ -48,6 +52,61 @@ describe("Functions gRPC support surface", () => { expect(completed?.getResult()?.getValue()).toBe('"done"'); }); + it("passes host features through a single orchestration request", async () => { + const featureKey = createMiddlewareFeature<{ invocationId: string }>("invocation"); + const features = new MiddlewareFeatures().set(featureKey, { invocationId: "host-123" }); + const worker = new TaskHubGrpcWorker({ logger: new NoOpLogger() }); + worker.useOrchestrationMiddleware(async (context, next) => { + expect(context.features.get(featureKey)?.invocationId).toBe("host-123"); + await next(context); + }); + const orchestrator: TOrchestrator = async () => "done"; + const name = worker.addOrchestrator(orchestrator); + const request = new pb.OrchestratorRequest(); + request.setInstanceid(TEST_INSTANCE_ID); + request.setNeweventsList([newOrchestratorStartedEvent(), newExecutionStartedEvent(name, TEST_INSTANCE_ID)]); + + const response = await worker.processOrchestratorRequest(request.serializeBinary(), features); + + expect(pb.OrchestratorResponse.deserializeBinary(response).getActionsList()).toHaveLength(1); + }); + + it("processes activity middleware with host features without the gRPC worker loop", async () => { + const featureKey = createMiddlewareFeature<{ invocationId: string }>("invocation"); + const features = new MiddlewareFeatures().set(featureKey, { invocationId: "host-456" }); + const worker = new TaskHubGrpcWorker({ logger: new NoOpLogger() }); + worker.useActivityMiddleware(async (context, next) => { + expect(context.features.get(featureKey)?.invocationId).toBe("host-456"); + await next(context); + }); + worker.addNamedActivity("activity", (_context: ActivityContext, input: unknown) => input); + const request = new pb.ActivityRequest(); + request.setName("activity"); + request.setTaskid(1); + const orchestrationInstance = new pb.OrchestrationInstance(); + orchestrationInstance.setInstanceid(TEST_INSTANCE_ID); + request.setOrchestrationinstance(orchestrationInstance); + const input = new StringValue(); + input.setValue('"input"'); + request.setInput(input); + + const response = await worker.processActivityRequest(request.serializeBinary(), features); + + expect(pb.ActivityResponse.deserializeBinary(response).getResult()?.getValue()).toBe('"input"'); + }); + + it("locks middleware registration while the worker is running", () => { + const worker = new TaskHubGrpcWorker({ logger: new NoOpLogger() }); + (worker as any)._isRunning = true; + + expect(() => worker.useOrchestrationMiddleware(async (context, next) => next(context))).toThrow( + "Cannot add orchestration middleware while worker is running.", + ); + expect(() => worker.useActivityMiddleware(async (context, next) => next(context))).toThrow( + "Cannot add activity middleware while worker is running.", + ); + }); + it("throws a distinct abandon error when versioning rejects a mismatched orchestration work item", async () => { // Configure the worker with Strict version matching and the Reject failure strategy. The work // item below carries no orchestration version, which does not match the worker's "2.0.0", so diff --git a/packages/durabletask-js/test/middleware.spec.ts b/packages/durabletask-js/test/middleware.spec.ts new file mode 100644 index 00000000..901e764b --- /dev/null +++ b/packages/durabletask-js/test/middleware.spec.ts @@ -0,0 +1,540 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { ActivityContext } from "../src/task/context/activity-context"; +import { OrchestrationContext } from "../src/task/context/orchestration-context"; +import { TOrchestrator } from "../src/types/orchestrator.type"; +import { NoOpLogger } from "../src/types/logger.type"; +import { + ActivityMiddleware, + MiddlewareFeatures, + OrchestrationMiddleware, + createMiddlewareFeature, +} from "../src/worker/middleware"; +import { ActivityExecutor } from "../src/worker/activity-executor"; +import { OrchestrationExecutor, OrchestrationExecutionResult } from "../src/worker/orchestration-executor"; +import { Registry } from "../src/worker/registry"; +import * as pb from "../src/proto/orchestrator_service_pb"; +import { + getStringValue, + newExecutionStartedEvent, + newOrchestratorStartedEvent, + newTaskCompletedEvent, + newTaskScheduledEvent, +} from "../src/utils/pb-helper.util"; + +const testLogger = new NoOpLogger(); +const instanceId = "middleware-instance"; + +describe("MiddlewareFeatures", () => { + it("stores host objects by typed symbol without serialization", () => { + const invocationFeature = createMiddlewareFeature<{ invocationId: string }>("invocation"); + const features = new MiddlewareFeatures(); + const hostObject = { invocationId: "host-123" }; + + features.set(invocationFeature, hostObject); + + expect(features.get(invocationFeature)).toBe(hostObject); + expect(JSON.stringify(features)).toBe("{}"); + expect(features.delete(invocationFeature)).toBe(true); + expect(features.get(invocationFeature)).toBeUndefined(); + }); +}); + +describe("Activity middleware", () => { + it("runs in registration order and exposes populated context, features, and result", async () => { + const calls: string[] = []; + const featureKey = createMiddlewareFeature<{ value: string }>("host"); + const features = new MiddlewareFeatures().set(featureKey, { value: "feature-value" }); + let capturedContext: Parameters[0] | undefined; + const first: ActivityMiddleware = async (context, next) => { + capturedContext = context; + calls.push("first-before"); + expect(context.result).toBeUndefined(); + await next(context); + expect(context.result).toEqual({ output: 42 }); + calls.push("first-after"); + }; + const second: ActivityMiddleware = async (context, next) => { + calls.push("second-before"); + expect(context.features.get(featureKey)?.value).toBe("feature-value"); + await next(context); + calls.push("second-after"); + }; + const registry = new Registry(); + const name = "activity"; + registry.addNamedActivity(name, (_context: ActivityContext, input: unknown) => { + calls.push("body"); + return { output: input }; + }); + const executor = new ActivityExecutor(registry, testLogger, [first, second]); + const rawInput = "42"; + + const result = await executor.execute(instanceId, name, 7, rawInput, { + version: "v2", + tags: { tenant: "contoso" }, + features, + }); + + expect(result).toBe('{"output":42}'); + expect(calls).toEqual(["first-before", "second-before", "body", "second-after", "first-after"]); + expect(capturedContext).toMatchObject({ + name, + instanceId, + taskId: 7, + version: "v2", + tags: { tenant: "contoso" }, + input: 42, + rawInput, + }); + expect(capturedContext?.activityContext.orchestrationId).toBe(instanceId); + expect(Object.isFrozen(capturedContext?.tags)).toBe(true); + }); + + it("allows explicit short-circuiting, including an undefined result", async () => { + const registry = new Registry(); + const body = jest.fn(() => "body"); + const name = "activity"; + registry.addNamedActivity(name, body); + const middleware: ActivityMiddleware = async (context) => { + context.setResult(undefined); + }; + const executor = new ActivityExecutor(registry, testLogger, [middleware]); + + await expect(executor.execute(instanceId, name, 1, undefined)).resolves.toBeUndefined(); + expect(body).not.toHaveBeenCalled(); + }); + + it("allows middleware to inspect and replace the result after next", async () => { + const registry = new Registry(); + const name = "activity"; + registry.addNamedActivity(name, () => "unredacted"); + const middleware: ActivityMiddleware = async (context, next) => { + await next(context); + expect(context.result).toBe("unredacted"); + context.setResult("redacted"); + }; + const executor = new ActivityExecutor(registry, testLogger, [middleware]); + + await expect(executor.execute(instanceId, name, 1)).resolves.toBe('"redacted"'); + }); + + it("rejects successful middleware that neither calls next nor sets a result", async () => { + const registry = new Registry(); + const name = "activity"; + registry.addNamedActivity(name, () => "body"); + const middleware: ActivityMiddleware = async () => {}; + const executor = new ActivityExecutor(registry, testLogger, [middleware]); + + await expect(executor.execute(instanceId, name, 1)).rejects.toThrow( + "Activity middleware must call next exactly once or call setResult", + ); + }); + + it("rejects duplicate next calls without invoking the body twice", async () => { + const registry = new Registry(); + const body = jest.fn(() => "body"); + const name = "activity"; + registry.addNamedActivity(name, body); + const middleware: ActivityMiddleware = async (context, next) => { + await next(context); + await next(context); + }; + const executor = new ActivityExecutor(registry, testLogger, [middleware]); + + await expect(executor.execute(instanceId, name, 1)).rejects.toThrow( + "Activity middleware must call next at most once", + ); + expect(body).toHaveBeenCalledTimes(1); + }); + + it("rejects an ignored duplicate next call without creating an unhandled rejection", async () => { + const registry = new Registry(); + const name = "activity"; + registry.addNamedActivity(name, () => "body"); + const middleware: ActivityMiddleware = async (context, next) => { + await next(context); + void next(context); + }; + const executor = new ActivityExecutor(registry, testLogger, [middleware]); + + await expect(executor.execute(instanceId, name, 1)).rejects.toThrow( + "Activity middleware must call next at most once", + ); + }); + + it("observes rejection throughout an ignored duplicate next chain", async () => { + const registry = new Registry(); + const name = "activity"; + registry.addNamedActivity(name, () => "body"); + const middleware: ActivityMiddleware = async (context, next) => { + await next(context); + void next(context) + .then(() => {}) + .finally(() => {}); + }; + const executor = new ActivityExecutor(registry, testLogger, [middleware]); + + await expect(executor.execute(instanceId, name, 1)).rejects.toThrow( + "Activity middleware must call next at most once", + ); + }); + + it("rejects calling next after setResult", async () => { + const registry = new Registry(); + const body = jest.fn(() => "body"); + const name = "activity"; + registry.addNamedActivity(name, body); + const middleware: ActivityMiddleware = async (context, next) => { + context.setResult("short-circuit"); + await next(context); + }; + const executor = new ActivityExecutor(registry, testLogger, [middleware]); + + await expect(executor.execute(instanceId, name, 1)).rejects.toThrow( + "Activity middleware cannot call next after setResult", + ); + expect(body).not.toHaveBeenCalled(); + }); + + it("does not allow outer middleware to swallow a nested contract violation", async () => { + const registry = new Registry(); + const name = "activity"; + registry.addNamedActivity(name, () => "body"); + const outer: ActivityMiddleware = async (context, next) => { + try { + await next(context); + } catch { + // Contract violations remain fatal even when user middleware catches them. + } + }; + const invalidInner: ActivityMiddleware = async () => {}; + const executor = new ActivityExecutor(registry, testLogger, [outer, invalidInner]); + + await expect(executor.execute(instanceId, name, 1)).rejects.toThrow( + "Activity middleware must call next exactly once or call setResult", + ); + }); + + it("rejects middleware that returns before next completes", async () => { + const registry = new Registry(); + const name = "activity"; + registry.addNamedActivity(name, async () => "body"); + const middleware: ActivityMiddleware = async (context, next) => { + void next(context); + }; + const executor = new ActivityExecutor(registry, testLogger, [middleware]); + + await expect(executor.execute(instanceId, name, 1)).rejects.toThrow( + "Activity middleware must not return before next(context) completes", + ); + }); + + it("exposes activity failure after next and propagates it", async () => { + const expected = new Error("activity failed"); + const registry = new Registry(); + const name = "activity"; + registry.addNamedActivity(name, () => { + throw expected; + }); + const middleware: ActivityMiddleware = async (context, next) => { + try { + await next(context); + } catch (error) { + expect(error).toBe(expected); + expect(context.failure).toBe(expected); + throw error; + } + }; + const executor = new ActivityExecutor(registry, testLogger, [middleware]); + + await expect(executor.execute(instanceId, name, 1)).rejects.toBe(expected); + }); + + it("propagates activity middleware failures", async () => { + const expected = new Error("middleware failed"); + const registry = new Registry(); + const name = "activity"; + registry.addNamedActivity(name, () => "body"); + const middleware: ActivityMiddleware = async () => { + throw expected; + }; + const executor = new ActivityExecutor(registry, testLogger, [middleware]); + + await expect(executor.execute(instanceId, name, 1)).rejects.toBe(expected); + }); +}); + +describe("Orchestration middleware", () => { + it("runs in registration order and exposes replay metadata, features, and result", async () => { + const calls: string[] = []; + const featureKey = createMiddlewareFeature<{ invocationId: string }>("host"); + const features = new MiddlewareFeatures().set(featureKey, { invocationId: "host-123" }); + let capturedContext: Parameters[0] | undefined; + const first: OrchestrationMiddleware = async (context, next) => { + capturedContext = context; + calls.push("first-before"); + expect(context.result).toBeUndefined(); + await next(context); + expect(context.result).toBe("done"); + calls.push("first-after"); + }; + const second: OrchestrationMiddleware = async (context, next) => { + calls.push("second-before"); + expect(context.features.get(featureKey)?.invocationId).toBe("host-123"); + await next(context); + calls.push("second-after"); + }; + const orchestrator: TOrchestrator = async (_context: OrchestrationContext, input: unknown) => { + calls.push("body"); + return input; + }; + const registry = new Registry(); + registry.addNamedOrchestrator("orchestrator", orchestrator); + const executionStarted = newExecutionStartedEvent( + "orchestrator", + instanceId, + '"done"', + { name: "parent", instanceId: "parent-instance", taskScheduledId: 3 }, + "execution-id", + ); + const started = executionStarted.getExecutionstarted()!; + started.setVersion(getStringValue("v3")); + started.getTagsMap().set("tenant", "contoso"); + const oldEvents = [newOrchestratorStartedEvent(), executionStarted]; + const executor = new OrchestrationExecutor(registry, testLogger, [first, second]); + + const result = await executor.execute( + instanceId, + oldEvents, + [newOrchestratorStartedEvent()], + "execution-id", + features, + ); + + expectCompleteResult(result, '"done"'); + expect(calls).toEqual(["first-before", "second-before", "body", "second-after", "first-after"]); + expect(capturedContext).toMatchObject({ + name: "orchestrator", + instanceId, + version: "v3", + parent: { name: "parent", instanceId: "parent-instance", taskScheduledId: 3 }, + tags: { tenant: "contoso" }, + input: "done", + rawInput: '"done"', + isReplaying: true, + }); + expect(capturedContext?.orchestrationContext.instanceId).toBe(instanceId); + expect(Object.isFrozen(capturedContext?.tags)).toBe(true); + }); + + it("rejects successful middleware that does not call next", async () => { + const middleware: OrchestrationMiddleware = async () => {}; + + const result = await executeOrchestration([middleware]); + + expectFailure(result, "Orchestration middleware must call next exactly once"); + }); + + it("rejects duplicate next calls without invoking the body twice", async () => { + const body = jest.fn(async () => "done"); + const middleware: OrchestrationMiddleware = async (context, next) => { + await next(context); + await next(context); + }; + + const result = await executeOrchestration([middleware], body); + + expectFailure(result, "Orchestration middleware must call next exactly once"); + expect(body).toHaveBeenCalledTimes(1); + }); + + it("rejects an ignored duplicate orchestration next call", async () => { + const middleware: OrchestrationMiddleware = async (context, next) => { + await next(context); + void next(context); + }; + + const result = await executeOrchestration([middleware]); + + expectFailure(result, "Orchestration middleware must call next exactly once"); + }); + + it("observes rejection from an ignored duplicate next finally-chain", async () => { + const middleware: OrchestrationMiddleware = async (context, next) => { + await next(context); + void next(context).finally(() => {}); + }; + + const result = await executeOrchestration([middleware]); + + expectFailure(result, "Orchestration middleware must call next exactly once"); + }); + + it("exposes orchestration failure after next", async () => { + const expected = new Error("orchestration failed"); + const middleware: OrchestrationMiddleware = async (context, next) => { + await next(context); + expect(context.failure).toBe(expected); + }; + + const result = await executeOrchestration([middleware], async () => { + throw expected; + }); + + expectFailure(result, expected.message); + }); + + it("routes orchestration middleware failures through durable failure actions", async () => { + const expected = new Error("middleware failed"); + const middleware: OrchestrationMiddleware = async () => { + throw expected; + }; + + const result = await executeOrchestration([middleware]); + + expectFailure(result, expected.message); + }); + + it("keeps post-next middleware suspended until the orchestration completes", async () => { + const calls: string[] = []; + const activity = (_context: ActivityContext) => "activity-result"; + const orchestrator: TOrchestrator = async function* (context: OrchestrationContext): any { + calls.push("body"); + return yield context.callActivity(activity); + }; + const middleware: OrchestrationMiddleware = async (context, next) => { + calls.push("before"); + await next(context); + calls.push(`after:${context.result}`); + }; + const registry = new Registry(); + const name = registry.addOrchestrator(orchestrator); + + const firstResult = await new OrchestrationExecutor(registry, testLogger, [middleware]).execute( + instanceId, + [], + [newOrchestratorStartedEvent(), newExecutionStartedEvent(name, instanceId)], + ); + + expect(firstResult.actions).toHaveLength(1); + expect(calls).toEqual(["before", "body"]); + + const secondResult = await new OrchestrationExecutor(registry, testLogger, [middleware]).execute( + instanceId, + [ + newOrchestratorStartedEvent(), + newExecutionStartedEvent(name, instanceId), + newTaskScheduledEvent(1, activity.name), + ], + [newTaskCompletedEvent(1, '"activity-result"')], + ); + + expectCompleteResult(secondResult, '"activity-result"'); + expect(calls).toEqual(["before", "body", "before", "body", "after:activity-result"]); + }); + + it("replays durable actions created before next in deterministic sequence", async () => { + const middlewareActivity = (_context: ActivityContext) => "middleware"; + const bodyActivity = (_context: ActivityContext) => "body"; + const orchestrator: TOrchestrator = async function* (context: OrchestrationContext): any { + return yield context.callActivity(bodyActivity); + }; + const middleware: OrchestrationMiddleware = async (context, next) => { + context.orchestrationContext.callActivity(middlewareActivity); + await next(context); + }; + const registry = new Registry(); + const name = registry.addOrchestrator(orchestrator); + + const firstResult = await new OrchestrationExecutor(registry, testLogger, [middleware]).execute( + instanceId, + [], + [newOrchestratorStartedEvent(), newExecutionStartedEvent(name, instanceId)], + ); + + expect(firstResult.actions.map((action) => action.getScheduletask()?.getName())).toEqual([ + middlewareActivity.name, + bodyActivity.name, + ]); + + const secondResult = await new OrchestrationExecutor(registry, testLogger, [middleware]).execute( + instanceId, + [ + newOrchestratorStartedEvent(), + newExecutionStartedEvent(name, instanceId), + newTaskScheduledEvent(1, middlewareActivity.name), + newTaskScheduledEvent(2, bodyActivity.name), + newTaskCompletedEvent(1, '"middleware"'), + ], + [newTaskCompletedEvent(2, '"body"')], + ); + + expectCompleteResult(secondResult, '"body"'); + }); + + it("does not allow outer middleware to swallow a nested next-call violation", async () => { + const outer: OrchestrationMiddleware = async (context, next) => { + try { + await next(context); + } catch { + // Contract violations remain fatal even when user middleware catches them. + } + }; + const invalidInner: OrchestrationMiddleware = async () => {}; + + const result = await executeOrchestration([outer, invalidInner]); + + expectFailure(result, "Orchestration middleware must call next exactly once"); + }); + + it("rejects orchestration middleware that returns before next completes", async () => { + const middleware: OrchestrationMiddleware = async (context, next) => { + void next(context); + }; + + const result = await executeOrchestration([middleware]); + + expectFailure(result, "Orchestration middleware must not return before next(context) completes"); + }); + + it("quiesces a started but unawaited orchestration next call before failing", async () => { + const activity = (_context: ActivityContext) => "body"; + const orchestrator: TOrchestrator = async function* (context: OrchestrationContext): any { + return yield context.callActivity(activity); + }; + const middleware: OrchestrationMiddleware = async (context, next) => { + void next(context).then(() => {}); + }; + + const result = await executeOrchestration([middleware], orchestrator); + + expectFailure(result, "Orchestration middleware must not return before next(context) completes"); + }); +}); + +async function executeOrchestration( + middleware: OrchestrationMiddleware[], + orchestrator: TOrchestrator = async () => "done", +): Promise { + const registry = new Registry(); + registry.addNamedOrchestrator("orchestrator", orchestrator); + const executor = new OrchestrationExecutor(registry, testLogger, middleware); + return executor.execute( + instanceId, + [], + [newOrchestratorStartedEvent(), newExecutionStartedEvent("orchestrator", instanceId)], + ); +} + +function expectCompleteResult(result: OrchestrationExecutionResult, expected: string): void { + const complete = result.actions.find((action) => action.hasCompleteorchestration())?.getCompleteorchestration(); + expect(complete?.getOrchestrationstatus()).toBe(pb.OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED); + expect(complete?.getResult()?.getValue()).toBe(expected); +} + +function expectFailure(result: OrchestrationExecutionResult, expectedMessage: string): void { + const complete = result.actions.find((action) => action.hasCompleteorchestration())?.getCompleteorchestration(); + expect(complete?.getOrchestrationstatus()).toBe(pb.OrchestrationStatus.ORCHESTRATION_STATUS_FAILED); + expect(complete?.getFailuredetails()?.getErrormessage()).toContain(expectedMessage); +} diff --git a/packages/durabletask-js/test/test-worker-middleware.spec.ts b/packages/durabletask-js/test/test-worker-middleware.spec.ts new file mode 100644 index 00000000..f38ea09a --- /dev/null +++ b/packages/durabletask-js/test/test-worker-middleware.spec.ts @@ -0,0 +1,63 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { InMemoryOrchestrationBackend } from "../src/testing/in-memory-backend"; +import { TestOrchestrationClient } from "../src/testing/test-client"; +import { TestOrchestrationWorker } from "../src/testing/test-worker"; +import { ActivityContext } from "../src/task/context/activity-context"; +import { OrchestrationContext } from "../src/task/context/orchestration-context"; +import { ActivityMiddlewareContext } from "../src/worker/middleware"; + +describe("TestOrchestrationWorker middleware registration", () => { + it("returns the worker for chaining and locks registration after start", async () => { + const worker = new TestOrchestrationWorker(new InMemoryOrchestrationBackend()); + const orchestrationMiddleware = jest.fn(async (context, next) => next(context)); + const activityMiddleware = jest.fn(async (context, next) => next(context)); + + expect(worker.useOrchestrationMiddleware(orchestrationMiddleware)).toBe(worker); + expect(worker.useActivityMiddleware(activityMiddleware)).toBe(worker); + + await worker.start(); + try { + expect(() => worker.useOrchestrationMiddleware(orchestrationMiddleware)).toThrow( + "Cannot add orchestration middleware while worker is running.", + ); + expect(() => worker.useActivityMiddleware(activityMiddleware)).toThrow( + "Cannot add activity middleware while worker is running.", + ); + } finally { + await worker.stop(); + } + }); + + it("preserves scheduled activity version and tags in middleware context", async () => { + const backend = new InMemoryOrchestrationBackend(); + const client = new TestOrchestrationClient(backend); + const worker = new TestOrchestrationWorker(backend); + let activityContext: ActivityMiddlewareContext | undefined; + const activity = (_context: ActivityContext, input: string) => input; + const orchestrator = async function* (context: OrchestrationContext): any { + return yield context.callActivity(activity, "input", { + version: "v2", + tags: { tenant: "contoso" }, + }); + }; + worker.useActivityMiddleware(async (context, next) => { + activityContext = context; + await next(context); + }); + worker.addActivity(activity); + worker.addOrchestrator(orchestrator); + + await worker.start(); + try { + const id = await client.scheduleNewOrchestration(orchestrator); + await client.waitForOrchestrationCompletion(id, true, 5); + } finally { + await worker.stop(); + } + + expect(activityContext?.version).toBe("v2"); + expect(activityContext?.tags).toEqual({ tenant: "contoso" }); + }); +}); From 9d59cc69ca65475c3a26fe0aa87fc1159dfacf1a Mon Sep 17 00:00:00 2001 From: wangbill Date: Wed, 5 Aug 2026 12:11:51 -0700 Subject: [PATCH 2/4] fix: preserve middleware failures Record downstream failures before rethrowing so outer middleware cannot silently swallow them. Allow activity recovery only through explicit setResult and keep orchestration failures fatal. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: f605785d-5c91-459e-829c-c32c158f907b --- README.md | 2 +- .../durabletask-js/src/worker/middleware.ts | 60 ++++- .../src/worker/orchestration-executor.ts | 6 +- .../durabletask-js/test/middleware.spec.ts | 217 ++++++++++++++++++ 4 files changed, 275 insertions(+), 10 deletions(-) diff --git a/README.md b/README.md index 879d50b8..89819ba4 100644 --- a/README.md +++ b/README.md @@ -63,7 +63,7 @@ worker worker.addActivity(myActivity); ``` -Orchestration middleware participates in generator replay: code before `next(context)` re-executes on each replay pass, while `next` remains suspended with the orchestrator and code after it unwinds only when the orchestration reaches a terminal result or failure. Orchestration middleware must call `next(context)` exactly once and must not return before it completes. It cannot short-circuit or replace the orchestration result. Activity middleware must either call `next(context)` once without returning before it completes, or explicitly short-circuit with `context.setResult(...)`; calling `next` after `setResult` is invalid, while calling `setResult` after `next` may replace a result. Missing, duplicate, and incomplete `next` calls are rejected. In normal `async` middleware, use `await next(context)` as shown above. +Orchestration middleware participates in generator replay: code before `next(context)` re-executes on each replay pass, while `next` remains suspended with the orchestrator and code after it unwinds only when the orchestration reaches a terminal result or failure. Orchestration middleware must call `next(context)` exactly once and must not return before it completes. It cannot short-circuit, replace the orchestration result, or suppress a downstream failure by catching it. Activity middleware must either call `next(context)` once without returning before it completes, or explicitly short-circuit with `context.setResult(...)`; calling `next` after `setResult` is invalid, while calling `setResult` after `next` may replace a result or explicitly recover from a caught downstream failure. Missing, duplicate, and incomplete `next` calls are rejected. In normal `async` middleware, use `await next(context)` as shown above. Orchestration middleware follows the same determinism rules as orchestrator code. Do not await nondurable promises or use wall-clock time, random values, mutable process state, file/network I/O, or host state that can change across replay. Use durable context APIs and `createReplaySafeLogger()` instead. Put I/O in activities or activity middleware. diff --git a/packages/durabletask-js/src/worker/middleware.ts b/packages/durabletask-js/src/worker/middleware.ts index e9fa55ae..eb4fd6da 100644 --- a/packages/durabletask-js/src/worker/middleware.ts +++ b/packages/durabletask-js/src/worker/middleware.ts @@ -184,6 +184,7 @@ export async function runOrchestrationMiddleware( suspended?: Promise, ): Promise<"completed" | "suspended"> { const validation = new MiddlewareValidation(); + let downstreamFailure: Error | undefined; const invoke = async (index: number): Promise => { if (index === middleware.length) { @@ -206,6 +207,11 @@ export async function runOrchestrationMiddleware( nextStarted = true; try { await invoke(index + 1); + } catch (error: unknown) { + const failure = toError(error); + downstreamFailure = combineFailures(downstreamFailure, failure); + context.setExecutionResult(context.result, combineFailures(context.failure, failure)); + throw failure; } finally { nextSettled = true; } @@ -237,11 +243,13 @@ export async function runOrchestrationMiddleware( if (outcome === "suspended") { await Promise.resolve(); } - validation.throwIfFailed(); + const failure = combineFailures(context.failure, downstreamFailure, validation.failure); + if (failure) { + throw failure; + } return outcome; } catch (error: unknown) { - validation.throwIfFailed(); - throw error; + throw combineFailures(context.failure, downstreamFailure, toError(error), validation.failure) ?? toError(error); } } @@ -287,6 +295,10 @@ export async function runActivityMiddleware( nextStarted = true; try { await invoke(index + 1); + } catch (error: unknown) { + const failure = toError(error); + context.setFailure(combineFailures(context.failure, failure) ?? failure); + throw failure; } finally { nextSettled = true; } @@ -308,17 +320,51 @@ export async function runActivityMiddleware( try { await invoke(0); - validation.throwIfFailed(); - if (context.failure) { - throw context.failure; + const failure = combineFailures(context.failure, validation.failure); + if (failure) { + throw failure; } } catch (error: unknown) { - const failure = validation.failure ?? (error instanceof Error ? error : new Error(String(error))); + const failure = combineFailures(context.failure, toError(error), validation.failure) ?? toError(error); context.setFailure(failure); throw failure; } } +function combineFailures(...failures: Array): Error | undefined { + const unique = failures.filter( + (failure, index): failure is Error => failure !== undefined && failures.indexOf(failure) === index, + ); + const distinct = unique.filter( + (failure) => !unique.some((other) => other !== failure && containsFailure(other, failure)), + ); + if (distinct.length === 0) { + return undefined; + } + if (distinct.length === 1) { + return distinct[0]; + } + + return new AggregateError( + distinct, + `Multiple middleware failures: ${distinct.map((failure) => failure.message).join("; ")}`, + ); +} + +function containsFailure(container: Error, candidate: Error): boolean { + return ( + container === candidate || + (container instanceof AggregateError && + container.errors.some( + (nested: unknown) => nested instanceof Error && containsFailure(nested, candidate), + )) + ); +} + +function toError(error: unknown): Error { + return error instanceof Error ? error : new Error(String(error)); +} + function copyTags(tags: Readonly> | undefined): Readonly> | undefined { return tags ? Object.freeze({ ...tags }) : undefined; } diff --git a/packages/durabletask-js/src/worker/orchestration-executor.ts b/packages/durabletask-js/src/worker/orchestration-executor.ts index 6189b5af..566b4e98 100644 --- a/packages/durabletask-js/src/worker/orchestration-executor.ts +++ b/packages/durabletask-js/src/worker/orchestration-executor.ts @@ -104,7 +104,9 @@ export class OrchestrationExecutor { const allEvents = [...oldEvents, ...newEvents]; const executionStarted = allEvents.find((event) => event.hasExecutionstarted())?.getExecutionstarted(); - const initialOrchestratorStarted = allEvents.find((event) => event.hasOrchestratorstarted()); + const currentOrchestratorStarted = + newEvents.find((event) => event.hasOrchestratorstarted()) ?? + [...oldEvents].reverse().find((event) => event.hasOrchestratorstarted()); const rawInput = isEmpty(executionStarted?.getInput()) ? undefined : executionStarted?.getInput()?.getValue(); const parentInstance = executionStarted?.getParentinstance(); const parentOrchestrationInstance = parentInstance?.getOrchestrationinstance(); @@ -117,7 +119,7 @@ export class OrchestrationExecutor { } : undefined; ctx._isReplaying = oldEvents.length > 0; - ctx._currentUtcDatetime = initialOrchestratorStarted?.getTimestamp()?.toDate() ?? ctx._currentUtcDatetime; + ctx._currentUtcDatetime = currentOrchestratorStarted?.getTimestamp()?.toDate() ?? ctx._currentUtcDatetime; this._orchestratorName = executionStarted?.getName() ?? "(unknown)"; try { diff --git a/packages/durabletask-js/test/middleware.spec.ts b/packages/durabletask-js/test/middleware.spec.ts index 901e764b..187d7c78 100644 --- a/packages/durabletask-js/test/middleware.spec.ts +++ b/packages/durabletask-js/test/middleware.spec.ts @@ -148,6 +148,81 @@ describe("Activity middleware", () => { expect(body).toHaveBeenCalledTimes(1); }); + it("preserves activity and duplicate-next failures together", async () => { + const bodyFailure = new Error("activity body failed"); + const registry = new Registry(); + const body = jest.fn(() => { + throw bodyFailure; + }); + const name = "activity"; + registry.addNamedActivity(name, body); + const middleware: ActivityMiddleware = async (context, next) => { + try { + await next(context); + } catch { + // Invoke next again to ensure the contract failure does not mask the body failure. + } + await next(context); + }; + const executor = new ActivityExecutor(registry, testLogger, [middleware]); + + const error = await executor.execute(instanceId, name, 1).catch((failure: unknown) => failure); + + expect(error).toBeInstanceOf(AggregateError); + if (!(error instanceof AggregateError)) { + throw new Error("Expected an AggregateError."); + } + expect(error.message).toContain(bodyFailure.message); + expect(error.message).toContain("Activity middleware must call next at most once"); + expect(error.errors).toEqual(expect.arrayContaining([bodyFailure])); + expect(body).toHaveBeenCalledTimes(1); + }); + + it("preserves an activity AggregateError when it is the only failure", async () => { + const expected = new AggregateError([new Error("child failure")], "activity aggregate failed"); + const registry = new Registry(); + const name = "activity"; + registry.addNamedActivity(name, () => { + throw expected; + }); + const executor = new ActivityExecutor(registry, testLogger); + + await expect(executor.execute(instanceId, name, 1)).rejects.toBe(expected); + }); + + it("preserves nested activity and duplicate-next failures when outer middleware catches", async () => { + const bodyFailure = new Error("nested activity body failed"); + const registry = new Registry(); + const body = jest.fn(() => { + throw bodyFailure; + }); + const name = "activity"; + registry.addNamedActivity(name, body); + const outer: ActivityMiddleware = async (context, next) => { + try { + await next(context); + } catch { + // Downstream failures remain recorded even when caught at this boundary. + } + }; + const inner: ActivityMiddleware = async (context, next) => { + try { + await next(context); + } catch { + // Trigger a contract failure without losing the activity failure. + } + await next(context); + }; + const executor = new ActivityExecutor(registry, testLogger, [outer, inner]); + + const error = await executor.execute(instanceId, name, 1).catch((failure: unknown) => failure); + + expect(error).toBeInstanceOf(AggregateError); + expect((error as Error).message).toContain(bodyFailure.message); + expect((error as Error).message).toContain("Activity middleware must call next at most once"); + expect(body).toHaveBeenCalledTimes(1); + }); + it("rejects an ignored duplicate next call without creating an unhandled rejection", async () => { const registry = new Registry(); const name = "activity"; @@ -216,6 +291,61 @@ describe("Activity middleware", () => { ); }); + it("does not allow outer middleware to swallow a nested activity failure", async () => { + const expected = new Error("inner activity middleware failed"); + let caught: unknown; + let observedFailure: Error | undefined; + const registry = new Registry(); + const body = jest.fn(() => "body"); + const name = "activity"; + registry.addNamedActivity(name, body); + const outer: ActivityMiddleware = async (context, next) => { + try { + await next(context); + } catch (error) { + caught = error; + observedFailure = context.failure; + } + }; + const failingInner: ActivityMiddleware = async () => { + throw expected; + }; + const executor = new ActivityExecutor(registry, testLogger, [outer, failingInner]); + + await expect(executor.execute(instanceId, name, 1)).rejects.toBe(expected); + expect(caught).toBe(expected); + expect(observedFailure).toBe(expected); + expect(body).not.toHaveBeenCalled(); + }); + + it("allows outer activity middleware to recover from a nested failure with setResult", async () => { + const expected = new Error("inner activity middleware failed"); + let caught: unknown; + let observedFailure: Error | undefined; + const registry = new Registry(); + const body = jest.fn(() => "body"); + const name = "activity"; + registry.addNamedActivity(name, body); + const outer: ActivityMiddleware = async (context, next) => { + try { + await next(context); + } catch (error) { + caught = error; + observedFailure = context.failure; + context.setResult("recovered"); + } + }; + const failingInner: ActivityMiddleware = async () => { + throw expected; + }; + const executor = new ActivityExecutor(registry, testLogger, [outer, failingInner]); + + await expect(executor.execute(instanceId, name, 1)).resolves.toBe('"recovered"'); + expect(caught).toBe(expected); + expect(observedFailure).toBe(expected); + expect(body).not.toHaveBeenCalled(); + }); + it("rejects middleware that returns before next completes", async () => { const registry = new Registry(); const name = "activity"; @@ -328,6 +458,27 @@ describe("Orchestration middleware", () => { expect(Object.isFrozen(capturedContext?.tags)).toBe(true); }); + it("exposes the current work-item timestamp before replay starts", async () => { + const previousTimestamp = new Date("2026-01-01T00:00:00.000Z"); + const currentTimestamp = new Date("2026-01-02T00:00:00.000Z"); + let observedTimestamp: Date | undefined; + const middleware: OrchestrationMiddleware = async (context, next) => { + observedTimestamp = context.orchestrationContext.currentUtcDateTime; + await next(context); + }; + const registry = new Registry(); + registry.addNamedOrchestrator("orchestrator", async () => "done"); + const executor = new OrchestrationExecutor(registry, testLogger, [middleware]); + + await executor.execute( + instanceId, + [newOrchestratorStartedEvent(previousTimestamp), newExecutionStartedEvent("orchestrator", instanceId)], + [newOrchestratorStartedEvent(currentTimestamp)], + ); + + expect(observedTimestamp).toEqual(currentTimestamp); + }); + it("rejects successful middleware that does not call next", async () => { const middleware: OrchestrationMiddleware = async () => {}; @@ -349,6 +500,47 @@ describe("Orchestration middleware", () => { expect(body).toHaveBeenCalledTimes(1); }); + it("preserves orchestration and duplicate-next failures together", async () => { + const bodyFailure = new Error("orchestration body failed"); + const body = jest.fn(async () => { + throw bodyFailure; + }); + const middleware: OrchestrationMiddleware = async (context, next) => { + await next(context); + await next(context); + }; + + const result = await executeOrchestration([middleware], body); + + expectFailure(result, bodyFailure.message); + expectFailure(result, "Orchestration middleware must call next exactly once"); + expect(body).toHaveBeenCalledTimes(1); + }); + + it("preserves nested orchestration and duplicate-next failures when outer middleware catches", async () => { + const bodyFailure = new Error("nested orchestration body failed"); + const body = jest.fn(async () => { + throw bodyFailure; + }); + const outer: OrchestrationMiddleware = async (context, next) => { + try { + await next(context); + } catch { + // Downstream failures remain fatal even when caught at this boundary. + } + }; + const inner: OrchestrationMiddleware = async (context, next) => { + await next(context); + await next(context); + }; + + const result = await executeOrchestration([outer, inner], body); + + expectFailure(result, bodyFailure.message); + expectFailure(result, "Orchestration middleware must call next exactly once"); + expect(body).toHaveBeenCalledTimes(1); + }); + it("rejects an ignored duplicate orchestration next call", async () => { const middleware: OrchestrationMiddleware = async (context, next) => { await next(context); @@ -488,6 +680,31 @@ describe("Orchestration middleware", () => { expectFailure(result, "Orchestration middleware must call next exactly once"); }); + it("does not allow outer middleware to swallow a nested orchestration failure", async () => { + const expected = new Error("inner orchestration middleware failed"); + let caught: unknown; + let observedFailure: Error | undefined; + const body = jest.fn(async () => "done"); + const outer: OrchestrationMiddleware = async (context, next) => { + try { + await next(context); + } catch (error) { + caught = error; + observedFailure = context.failure; + } + }; + const failingInner: OrchestrationMiddleware = async () => { + throw expected; + }; + + const result = await executeOrchestration([outer, failingInner], body); + + expectFailure(result, expected.message); + expect(caught).toBe(expected); + expect(observedFailure).toBe(expected); + expect(body).not.toHaveBeenCalled(); + }); + it("rejects orchestration middleware that returns before next completes", async () => { const middleware: OrchestrationMiddleware = async (context, next) => { void next(context); From 5190c681ab5d48d521f6975beb331c5be7d51669 Mon Sep 17 00:00:00 2001 From: wangbill Date: Wed, 5 Aug 2026 12:14:08 -0700 Subject: [PATCH 3/4] Format middleware failure handling Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: c0f06a45-0b8f-4a9e-88c9-e71d057b1f79 --- packages/durabletask-js/src/worker/middleware.ts | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/packages/durabletask-js/src/worker/middleware.ts b/packages/durabletask-js/src/worker/middleware.ts index eb4fd6da..7a828a53 100644 --- a/packages/durabletask-js/src/worker/middleware.ts +++ b/packages/durabletask-js/src/worker/middleware.ts @@ -355,9 +355,7 @@ function containsFailure(container: Error, candidate: Error): boolean { return ( container === candidate || (container instanceof AggregateError && - container.errors.some( - (nested: unknown) => nested instanceof Error && containsFailure(nested, candidate), - )) + container.errors.some((nested: unknown) => nested instanceof Error && containsFailure(nested, candidate))) ); } From 75f2cf5b7690110279c944c404ffcf04c360304d Mon Sep 17 00:00:00 2001 From: wangbill Date: Wed, 5 Aug 2026 12:21:33 -0700 Subject: [PATCH 4/4] Preserve middleware replay determinism Keep pre-next middleware on the original orchestration clock so deterministic values and actions reproduce before history replay. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: c0f06a45-0b8f-4a9e-88c9-e71d057b1f79 --- .../src/worker/orchestration-executor.ts | 6 +- .../durabletask-js/test/middleware.spec.ts | 64 +++++++++++++------ 2 files changed, 45 insertions(+), 25 deletions(-) diff --git a/packages/durabletask-js/src/worker/orchestration-executor.ts b/packages/durabletask-js/src/worker/orchestration-executor.ts index 566b4e98..6189b5af 100644 --- a/packages/durabletask-js/src/worker/orchestration-executor.ts +++ b/packages/durabletask-js/src/worker/orchestration-executor.ts @@ -104,9 +104,7 @@ export class OrchestrationExecutor { const allEvents = [...oldEvents, ...newEvents]; const executionStarted = allEvents.find((event) => event.hasExecutionstarted())?.getExecutionstarted(); - const currentOrchestratorStarted = - newEvents.find((event) => event.hasOrchestratorstarted()) ?? - [...oldEvents].reverse().find((event) => event.hasOrchestratorstarted()); + const initialOrchestratorStarted = allEvents.find((event) => event.hasOrchestratorstarted()); const rawInput = isEmpty(executionStarted?.getInput()) ? undefined : executionStarted?.getInput()?.getValue(); const parentInstance = executionStarted?.getParentinstance(); const parentOrchestrationInstance = parentInstance?.getOrchestrationinstance(); @@ -119,7 +117,7 @@ export class OrchestrationExecutor { } : undefined; ctx._isReplaying = oldEvents.length > 0; - ctx._currentUtcDatetime = currentOrchestratorStarted?.getTimestamp()?.toDate() ?? ctx._currentUtcDatetime; + ctx._currentUtcDatetime = initialOrchestratorStarted?.getTimestamp()?.toDate() ?? ctx._currentUtcDatetime; this._orchestratorName = executionStarted?.getName() ?? "(unknown)"; try { diff --git a/packages/durabletask-js/test/middleware.spec.ts b/packages/durabletask-js/test/middleware.spec.ts index 187d7c78..654d8b4a 100644 --- a/packages/durabletask-js/test/middleware.spec.ts +++ b/packages/durabletask-js/test/middleware.spec.ts @@ -458,27 +458,6 @@ describe("Orchestration middleware", () => { expect(Object.isFrozen(capturedContext?.tags)).toBe(true); }); - it("exposes the current work-item timestamp before replay starts", async () => { - const previousTimestamp = new Date("2026-01-01T00:00:00.000Z"); - const currentTimestamp = new Date("2026-01-02T00:00:00.000Z"); - let observedTimestamp: Date | undefined; - const middleware: OrchestrationMiddleware = async (context, next) => { - observedTimestamp = context.orchestrationContext.currentUtcDateTime; - await next(context); - }; - const registry = new Registry(); - registry.addNamedOrchestrator("orchestrator", async () => "done"); - const executor = new OrchestrationExecutor(registry, testLogger, [middleware]); - - await executor.execute( - instanceId, - [newOrchestratorStartedEvent(previousTimestamp), newExecutionStartedEvent("orchestrator", instanceId)], - [newOrchestratorStartedEvent(currentTimestamp)], - ); - - expect(observedTimestamp).toEqual(currentTimestamp); - }); - it("rejects successful middleware that does not call next", async () => { const middleware: OrchestrationMiddleware = async () => {}; @@ -665,6 +644,49 @@ describe("Orchestration middleware", () => { expectCompleteResult(secondResult, '"body"'); }); + it("replays pre-next deterministic values against the original orchestration clock", async () => { + const firstTimestamp = new Date("2026-01-01T00:00:00.000Z"); + const secondTimestamp = new Date("2026-01-02T00:00:00.000Z"); + const generatedGuids: string[] = []; + const middlewareActivity = (_context: ActivityContext) => "middleware"; + const bodyActivity = (_context: ActivityContext) => "body"; + const orchestrator: TOrchestrator = async function* (context: OrchestrationContext): any { + return yield context.callActivity(bodyActivity); + }; + const middleware: OrchestrationMiddleware = async (context, next) => { + const guid = context.orchestrationContext.newGuid(); + generatedGuids.push(guid); + context.orchestrationContext.callActivity(middlewareActivity, guid); + await next(context); + }; + const registry = new Registry(); + const name = registry.addOrchestrator(orchestrator); + + const firstResult = await new OrchestrationExecutor(registry, testLogger, [middleware]).execute( + instanceId, + [], + [newOrchestratorStartedEvent(firstTimestamp), newExecutionStartedEvent(name, instanceId)], + ); + + expect(firstResult.actions[0].getScheduletask()?.getInput()?.getValue()).toBe(JSON.stringify(generatedGuids[0])); + + const secondResult = await new OrchestrationExecutor(registry, testLogger, [middleware]).execute( + instanceId, + [ + newOrchestratorStartedEvent(firstTimestamp), + newExecutionStartedEvent(name, instanceId), + newTaskScheduledEvent(1, middlewareActivity.name), + newTaskScheduledEvent(2, bodyActivity.name), + newTaskCompletedEvent(1, '"middleware"'), + ], + [newOrchestratorStartedEvent(secondTimestamp), newTaskCompletedEvent(2, '"body"')], + ); + + expectCompleteResult(secondResult, '"body"'); + expect(generatedGuids).toHaveLength(2); + expect(generatedGuids[1]).toBe(generatedGuids[0]); + }); + it("does not allow outer middleware to swallow a nested next-call violation", async () => { const outer: OrchestrationMiddleware = async (context, next) => { try {