diff --git a/CHANGELOG.md b/CHANGELOG.md index b4508367..e7566596 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ### New +- Add first-class worker-level orchestration and activity middleware with typed contexts, nonserialized host features, validation, and in-process host integration. - Add an optional `newVersion` parameter to `OrchestrationContext.continueAsNew()` for version migrations. - Implement entity support in the in-memory testing backend ([#341](https://github.com/microsoft/durabletask-js/pull/341)) - Add the top-level `StartOrchestrationOptions.dedupeStatuses` option, `ValidDedupeStatuses`, and diff --git a/README.md b/README.md index 9c5b9ef9..ae466282 100644 --- a/README.md +++ b/README.md @@ -18,11 +18,79 @@ 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, 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. + +### 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. @@ -176,10 +244,7 @@ Long-running orchestrations can restart with fresh history and optionally move t orchestration version: ```typescript -const eternalOrchestrator: TOrchestrator = async function* ( - ctx: OrchestrationContext, - iteration: number, -): any { +const eternalOrchestrator: TOrchestrator = async function* (ctx: OrchestrationContext, iteration: number): any { yield ctx.callActivity(processIteration, iteration); ctx.continueAsNew(iteration + 1, true, "2.0.0"); }; 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 bdf3e6ee..ea45c24e 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 6937381a..80859b79 100644 --- a/packages/durabletask-js/src/testing/in-memory-backend.ts +++ b/packages/durabletask-js/src/testing/in-memory-backend.ts @@ -7,6 +7,7 @@ import { OrchestrationStatus as ClientOrchestrationStatus } from "../orchestrati import { ParentOrchestrationInstance } from "../types/parent-orchestration-instance.type"; import { StringValue } from "google-protobuf/google/protobuf/wrappers_pb"; import { randomUUID } from "crypto"; +import { mapToRecord } from "../utils/tags.util"; import { validateDedupeStatusesForReplacement } from "../orchestration/orchestration-id-reuse-policy"; import { OrchestrationAlreadyExistsError } from "../orchestration/exception/orchestration-already-exists-error"; @@ -43,6 +44,8 @@ export interface ActivityWorkItem { name: string; taskId: number; input?: string; + version?: string; + tags?: Record; completionToken: number; } @@ -1054,9 +1057,7 @@ export class InMemoryOrchestrationBackend { ?.getExecutionstarted() ?.getVersion() ?.getValue(); - const newVersion = completeAction.hasNewversion() - ? completeAction.getNewversion()?.getValue() - : currentVersion; + const newVersion = completeAction.hasNewversion() ? completeAction.getNewversion()?.getValue() : currentVersion; const carryoverEvents = completeAction.getCarryovereventsList(); // Cancel timers still pending from the previous iteration. Their timer IDs are @@ -1092,7 +1093,7 @@ export class InMemoryOrchestrationBackend { newInput, undefined, instance.executionId, - newVersion + newVersion, ); instance.pendingEvents = [orchestratorStarted, executionStarted, ...carryoverEvents]; @@ -1105,6 +1106,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); @@ -1122,6 +1125,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 dbd3b19b..7a9475c7 100644 --- a/packages/durabletask-js/src/testing/test-worker.ts +++ b/packages/durabletask-js/src/testing/test-worker.ts @@ -20,10 +20,11 @@ import { import { StringValue } from "google-protobuf/google/protobuf/wrappers_pb"; 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. @@ -34,6 +35,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(); @@ -103,6 +106,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. */ @@ -178,7 +203,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); @@ -200,11 +225,11 @@ export class TestOrchestrationWorker { * Processes a single activity work item. */ private async processActivity(workItem: ActivityWorkItem): Promise { - const { instanceId, executionId, name, taskId, input } = workItem; + const { instanceId, executionId, name, taskId, input, version, 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, executionId, 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..7a828a53 --- /dev/null +++ b/packages/durabletask-js/src/worker/middleware.ts @@ -0,0 +1,410 @@ +// 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(); + let downstreamFailure: Error | undefined; + + 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); + } catch (error: unknown) { + const failure = toError(error); + downstreamFailure = combineFailures(downstreamFailure, failure); + context.setExecutionResult(context.result, combineFailures(context.failure, failure)); + throw failure; + } 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(); + } + const failure = combineFailures(context.failure, downstreamFailure, validation.failure); + if (failure) { + throw failure; + } + return outcome; + } catch (error: unknown) { + throw combineFailures(context.failure, downstreamFailure, toError(error), validation.failure) ?? toError(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); + } catch (error: unknown) { + const failure = toError(error); + context.setFailure(combineFailures(context.failure, failure) ?? failure); + throw failure; + } 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); + const failure = combineFailures(context.failure, validation.failure); + if (failure) { + throw failure; + } + } catch (error: unknown) { + 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; +} + +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 39caea41..d24164bc 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; @@ -60,6 +57,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; @@ -223,6 +221,7 @@ export class RuntimeOrchestrationContext extends OrchestrationContext { // must be preserved and returned alongside the complete action this._result = result; + this._failure = undefined; let resultJson; @@ -248,6 +247,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 @@ -333,15 +333,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); @@ -419,7 +415,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"); @@ -490,10 +493,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, + }); } } @@ -645,15 +647,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; } @@ -662,13 +659,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; } @@ -749,10 +740,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. @@ -815,25 +803,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 @@ -890,18 +872,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."); } @@ -957,10 +932,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); } } @@ -970,12 +942,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; @@ -1009,9 +976,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; @@ -1100,4 +1065,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..654d8b4a --- /dev/null +++ b/packages/durabletask-js/test/middleware.spec.ts @@ -0,0 +1,779 @@ +// 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("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"; + 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("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"; + 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("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); + 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("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 { + 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("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); + }; + + 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" }); + }); +});