diff --git a/README.md b/README.md index 823f6de..17ffbf6 100644 --- a/README.md +++ b/README.md @@ -23,6 +23,19 @@ const entityResponseBytes = await worker.processEntityBatchRequest(entityBatchRe `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`. +## External payload transport + +Core clients and workers accept opt-in `payloads: { store }` options, where `store` implements +the exported `PayloadStore` contract. The same store must be configured on every participant. +The default threshold is 256 KiB of serialized UTF-8 bytes (inclusive), with a 10 MiB maximum. +Requests are externalized before sending; queries, history chunks and work items are hydrated +before conversion or execution. Empty payloads remain inline, and disabling the option preserves +the existing transport behavior. + +Azure-managed builders expose `.useExternalizedPayloads({ store })` and require core `>=0.4.1` +for the public payload types. This phase works with a custom/in-memory store and does not import +an Azure Blob provider or register cleanup tasks. Complete the six-phase feature stack before release. + ## Long durable timers `createTimer(Date | seconds)` has no SDK-imposed total-duration cap. Like the Python SDK, diff --git a/package-lock.json b/package-lock.json index e1d0443..2922319 100644 --- a/package-lock.json +++ b/package-lock.json @@ -7671,7 +7671,7 @@ }, "peerDependencies": { "@grpc/grpc-js": "^1.14.4", - "@microsoft/durabletask-js": ">=0.3.0" + "@microsoft/durabletask-js": ">=0.4.1" } }, "packages/durabletask-js-azuremanaged/node_modules/@types/node": { diff --git a/packages/durabletask-js-azuremanaged/CHANGELOG.md b/packages/durabletask-js-azuremanaged/CHANGELOG.md index 60244b3..a57bd6c 100644 --- a/packages/durabletask-js-azuremanaged/CHANGELOG.md +++ b/packages/durabletask-js-azuremanaged/CHANGELOG.md @@ -2,6 +2,9 @@ ### New +- Add `.useExternalizedPayloads()` to client and worker builders, preserving settings when + connection options change. The new public option types require core `>=0.4.1`. + Cleanup registration is deferred to the later worker phase; do not release partial feature phases. - Support the `ResourceId` token audience URI in connection strings and an optional final `resourceId` argument in the client/worker factory functions, alongside existing builder/options setters. Trim surrounding whitespace and trailing slashes, remove one case-insensitive `/.default` suffix, diff --git a/packages/durabletask-js-azuremanaged/package.json b/packages/durabletask-js-azuremanaged/package.json index 29f72a6..151e9a4 100644 --- a/packages/durabletask-js-azuremanaged/package.json +++ b/packages/durabletask-js-azuremanaged/package.json @@ -53,7 +53,7 @@ }, "peerDependencies": { "@grpc/grpc-js": "^1.14.4", - "@microsoft/durabletask-js": ">=0.3.0" + "@microsoft/durabletask-js": ">=0.4.1" }, "devDependencies": { "@types/jest": "^29.5.1", diff --git a/packages/durabletask-js-azuremanaged/src/client-builder.ts b/packages/durabletask-js-azuremanaged/src/client-builder.ts index eab641b..1c84851 100644 --- a/packages/durabletask-js-azuremanaged/src/client-builder.ts +++ b/packages/durabletask-js-azuremanaged/src/client-builder.ts @@ -5,7 +5,7 @@ import { TokenCredential } from "@azure/identity"; import * as grpc from "@grpc/grpc-js"; import { DurableTaskAzureManagedClientOptions } from "./options"; import { ClientRetryOptions } from "./retry-policy"; -import { TaskHubGrpcClient, Logger, ConsoleLogger } from "@microsoft/durabletask-js"; +import { TaskHubGrpcClient, Logger, ConsoleLogger, PayloadStorageOptions } from "@microsoft/durabletask-js"; /** * Builder for creating DurableTaskClient instances that connect to Azure-managed Durable Task service. @@ -15,6 +15,7 @@ export class DurableTaskAzureManagedClientBuilder { private _options: DurableTaskAzureManagedClientOptions; private _grpcChannelOptions: grpc.ChannelOptions = {}; private _logger: Logger = new ConsoleLogger(); + private _payloads?: PayloadStorageOptions; /** * Creates a new instance of DurableTaskAzureManagedClientBuilder. @@ -142,6 +143,12 @@ export class DurableTaskAzureManagedClientBuilder { return this; } + /** Enables transparent externalized payloads using an independently configured payload store. */ + useExternalizedPayloads(options: PayloadStorageOptions): DurableTaskAzureManagedClientBuilder { + this._payloads = options; + return this; + } + /** * Builds and returns a configured TaskHubGrpcClient. * @@ -164,14 +171,15 @@ export class DurableTaskAzureManagedClientBuilder { }; // Use the core TaskHubGrpcClient with custom channel credentials and per-call metadata. - return new TaskHubGrpcClient( + return new TaskHubGrpcClient({ hostAddress, - combinedOptions, - true, - channelCredentials, + options: combinedOptions, + useTLS: true, + credentials: channelCredentials, metadataGenerator, - this._logger, - ); + logger: this._logger, + payloads: this._payloads, + }); } } diff --git a/packages/durabletask-js-azuremanaged/src/worker-builder.ts b/packages/durabletask-js-azuremanaged/src/worker-builder.ts index 8a0b1a7..05e6141 100644 --- a/packages/durabletask-js-azuremanaged/src/worker-builder.ts +++ b/packages/durabletask-js-azuremanaged/src/worker-builder.ts @@ -15,6 +15,7 @@ import { ConsoleLogger, VersioningOptions, WorkItemFilters, + PayloadStorageOptions, } from "@microsoft/durabletask-js"; /** @@ -33,6 +34,7 @@ export class DurableTaskAzureManagedWorkerBuilder { private _channelRecreateFailureThreshold?: number; private _versioning?: VersioningOptions; private _workItemFilters?: WorkItemFilters | "auto"; + private _payloads?: PayloadStorageOptions; /** * Creates a new instance of DurableTaskAzureManagedWorkerBuilder. @@ -305,6 +307,12 @@ export class DurableTaskAzureManagedWorkerBuilder { return this; } + /** Enables payload hydration, externalization and the reserved auto-purge tasks. */ + useExternalizedPayloads(options: PayloadStorageOptions): DurableTaskAzureManagedWorkerBuilder { + this._payloads = options; + return this; + } + /** * Builds and returns a configured TaskHubGrpcWorker. * @@ -339,6 +347,7 @@ export class DurableTaskAzureManagedWorkerBuilder { channelRecreateFailureThreshold: this._channelRecreateFailureThreshold, versioning: this._versioning, workItemFilters: this._workItemFilters, + payloads: this._payloads, // DTS natively supports long timers, matching the Python Azure-managed worker. maximumTimerIntervalMs: null, }); diff --git a/packages/durabletask-js-azuremanaged/test/unit/payload-builders.spec.ts b/packages/durabletask-js-azuremanaged/test/unit/payload-builders.spec.ts new file mode 100644 index 0000000..d797512 --- /dev/null +++ b/packages/durabletask-js-azuremanaged/test/unit/payload-builders.spec.ts @@ -0,0 +1,46 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { PayloadStore, NoOpLogger } from "@microsoft/durabletask-js"; +import { DurableTaskAzureManagedClientBuilder, DurableTaskAzureManagedWorkerBuilder } from "../../src"; +import * as pb from "../../../durabletask-js/src/proto/orchestrator_service_pb"; + +class Store extends PayloadStore { + async upload(): Promise { + return "external:payload"; + } + async download(): Promise { + return '"resolved"'; + } + isKnownPayloadToken(value: string): boolean { + return value.startsWith("external:"); + } +} + +describe("Azure-managed payload configuration", () => { + it("preserves client payload settings when connection options are replaced", async () => { + const client = new DurableTaskAzureManagedClientBuilder() + .useExternalizedPayloads({ store: new Store(), thresholdBytes: 1 }) + .connectionString("Endpoint=http://localhost:8080;Authentication=None;TaskHub=payload-test") + .logger(new NoOpLogger()) + .build(); + expect(client["_payloads"]).toBeDefined(); + await client.stop(); + }); + it("preserves worker payload settings and business registrations", () => { + const worker = new DurableTaskAzureManagedWorkerBuilder() + .useExternalizedPayloads({ store: new Store() }) + .addNamedOrchestrator("business", () => undefined) + .connectionString("Endpoint=http://localhost:8080;Authentication=None;TaskHub=payload-test") + .useWorkItemFilters() + .build(); + const request = worker["_buildGetWorkItemsRequest"](); + expect(request.getCapabilitiesList()).toContain(pb.WorkerCapability.WORKER_CAPABILITY_LARGE_PAYLOADS); + expect( + request + .getWorkitemfilters()! + .getOrchestrationsList() + .map((f) => f.getName()), + ).toEqual(["business"]); + }); +}); diff --git a/packages/durabletask-js/src/client/client.ts b/packages/durabletask-js/src/client/client.ts index 54d3bb7..f71ba9f 100644 --- a/packages/durabletask-js/src/client/client.ts +++ b/packages/durabletask-js/src/client/client.ts @@ -37,6 +37,9 @@ import { EntityInstanceId } from "../entities/entity-instance-id"; import { EntityMetadata, createEntityMetadata, createEntityMetadataWithoutState } from "../entities/entity-metadata"; import { EntityQuery, normalizeInstanceIdPrefix } from "../entities/entity-query"; import { SignalEntityOptions } from "../entities/signal-entity-options"; +import { PayloadStorageOptions } from "../payloads/payload-store"; +import { PayloadProcessor } from "../payloads/payload-processor"; +import { readPayloadHistory, PayloadHydrationError } from "../payloads/history-stream"; import { CleanEntityStorageRequest, CleanEntityStorageResult, @@ -59,6 +62,8 @@ export { MetadataGenerator } from "../utils/grpc-helper.util"; * Options for creating a TaskHubGrpcClient. */ export interface TaskHubGrpcClientOptions { + /** Optional externalized payload storage. Disabled by default. */ + payloads?: PayloadStorageOptions; /** The host address to connect to. Defaults to "localhost:4001". */ hostAddress?: string; /** gRPC channel options. */ @@ -84,6 +89,8 @@ export class TaskHubGrpcClient { private _metadataGenerator?: MetadataGenerator; private _logger: Logger; private _defaultVersion?: string; + private readonly _payloads?: PayloadProcessor; + private readonly _payloadCancellation = new AbortController(); /** * Creates a new TaskHubGrpcClient instance. @@ -137,6 +144,7 @@ export class TaskHubGrpcClient { resolvedMetadataGenerator = hostAddressOrOptions.metadataGenerator; resolvedLogger = hostAddressOrOptions.logger; resolvedDefaultVersion = hostAddressOrOptions.defaultVersion; + if (hostAddressOrOptions.payloads) this._payloads = new PayloadProcessor(hostAddressOrOptions.payloads); } else { // Deprecated positional parameters constructor resolvedHostAddress = hostAddressOrOptions; @@ -154,6 +162,7 @@ export class TaskHubGrpcClient { } async stop(): Promise { + this._payloadCancellation.abort(); await this._stub.close(); // Brief pause to allow gRPC cleanup - this is a known issue with grpc-node @@ -161,6 +170,22 @@ export class TaskHubGrpcClient { await new Promise((resolve) => setTimeout(resolve, 1000)); } + private async _call( + method: Parameters>[0], + request: TReq, + metadataGenerator?: MetadataGenerator, + signal?: AbortSignal, + ): Promise { + if (!this._payloads) return callWithMetadata(method, request, metadataGenerator, signal); + const payloadSignal = signal + ? AbortSignal.any([signal, this._payloadCancellation.signal]) + : this._payloadCancellation.signal; + await this._payloads.externalize(request, payloadSignal); + const response = await callWithMetadata(method, request, metadataGenerator, payloadSignal); + await this._payloads.resolve(response, payloadSignal); + return response; + } + /** * Schedules a new orchestrator using the DurableTask client. * @@ -256,7 +281,7 @@ export class TaskHubGrpcClient { ClientLogs.schedulingOrchestration(this._logger, req.getInstanceid(), name, inputSizeInBytes); try { - const res = await callWithMetadata( + const res = await this._call( this._stub.startInstance.bind(this._stub), req, this._metadataGenerator, @@ -301,7 +326,7 @@ export class TaskHubGrpcClient { req.setInstanceid(instanceId); req.setGetinputsandoutputs(fetchPayloads); - const res = await callWithMetadata( + const res = await this._call( this._stub.getInstance.bind(this._stub), req, this._metadataGenerator, @@ -343,7 +368,7 @@ export class TaskHubGrpcClient { signal, `Timed out waiting for orchestration '${instanceId}' to start after ${timeout}s`, (waitSignal) => - callWithMetadata(this._stub.waitForInstanceStart.bind(this._stub), req, this._metadataGenerator, waitSignal), + this._call(this._stub.waitForInstanceStart.bind(this._stub), req, this._metadataGenerator, waitSignal), ); return newOrchestrationState(req.getInstanceid(), res); @@ -415,7 +440,7 @@ export class TaskHubGrpcClient { // Metadata failures must not enter the remote-deadline retry path. const metadata = this._metadataGenerator ? await this._metadataGenerator() : new grpc.Metadata(); try { - return await callWithMetadata( + return await this._call( this._stub.waitForInstanceCompletion.bind(this._stub), req, async () => metadata, @@ -498,7 +523,7 @@ export class TaskHubGrpcClient { ClientLogs.raisingEvent(this._logger, instanceId, eventName); try { - await callWithMetadata( + await this._call( this._stub.raiseEvent.bind(this._stub), req, this._metadataGenerator, @@ -565,7 +590,7 @@ export class TaskHubGrpcClient { ClientLogs.terminatingInstance(this._logger, instanceId); - await callWithMetadata( + await this._call( this._stub.terminateInstance.bind(this._stub), req, this._metadataGenerator, @@ -591,7 +616,7 @@ export class TaskHubGrpcClient { ClientLogs.suspendingInstance(this._logger, instanceId); - await callWithMetadata( + await this._call( this._stub.suspendInstance.bind(this._stub), req, this._metadataGenerator, @@ -617,7 +642,7 @@ export class TaskHubGrpcClient { ClientLogs.resumingInstance(this._logger, instanceId); - await callWithMetadata( + await this._call( this._stub.resumeInstance.bind(this._stub), req, this._metadataGenerator, @@ -656,7 +681,7 @@ export class TaskHubGrpcClient { ClientLogs.rewindingInstance(this._logger, instanceId, reason); try { - await callWithMetadata( + await this._call( this._stub.rewindInstance.bind(this._stub), req, this._metadataGenerator, @@ -721,7 +746,7 @@ export class TaskHubGrpcClient { ClientLogs.restartingInstance(this._logger, instanceId, restartWithNewInstanceId); try { - const res = await callWithMetadata( + const res = await this._call( this._stub.restartInstance.bind(this._stub), req, this._metadataGenerator, @@ -775,7 +800,7 @@ export class TaskHubGrpcClient { ClientLogs.purgingInstanceMetadata(this._logger, instanceId); - res = await callWithMetadata( + res = await this._call( this._stub.purgeInstances.bind(this._stub), req, this._metadataGenerator, @@ -811,7 +836,7 @@ export class TaskHubGrpcClient { runtimeStatusList.map(String).join(", "), ); - const callPromise = callWithMetadata( + const callPromise = this._call( this._stub.purgeInstances.bind(this._stub), req, this._metadataGenerator, @@ -915,7 +940,7 @@ export class TaskHubGrpcClient { req.setQuery(query); - const response = await callWithMetadata( + const response = await this._call( this._stub.queryInstances.bind(this._stub), req, this._metadataGenerator, @@ -1001,7 +1026,7 @@ export class TaskHubGrpcClient { req.setCompletedtimeto(timestamp); } - const response = await callWithMetadata( + const response = await this._call( this._stub.listInstanceIds.bind(this._stub), req, this._metadataGenerator, @@ -1049,6 +1074,26 @@ export class TaskHubGrpcClient { const metadata = this._metadataGenerator ? await this._metadataGenerator() : new grpc.Metadata(); const stream = this._stub.streamInstanceHistory(req, metadata); + if (this._payloads) { + try { + const events = await readPayloadHistory(stream, this._payloads, this._payloadCancellation.signal); + return events.map(convertProtoHistoryEvent).filter((event): event is HistoryEvent => event !== undefined); + } catch (error) { + if (error instanceof PayloadHydrationError) throw error.cause; + if (error instanceof Error && "code" in error) { + if (error.code === grpc.status.NOT_FOUND) return []; + if (error.code === grpc.status.CANCELLED) + throw new Error("The getOrchestrationHistory operation was canceled."); + if (error.code === grpc.status.INTERNAL) { + throw new Error( + `An error occurred while retrieving the history for orchestration with instanceId '${instanceId}'.`, + ); + } + } + throw error; + } + } + return new Promise((resolve, reject) => { const historyEvents: HistoryEvent[] = []; @@ -1147,7 +1192,7 @@ export class TaskHubGrpcClient { ClientLogs.signalingEntity(this._logger, id.toString(), operationName); try { - await callWithMetadata( + await this._call( this._stub.signalEntity.bind(this._stub), req, this._metadataGenerator, @@ -1183,7 +1228,7 @@ export class TaskHubGrpcClient { ClientLogs.gettingEntity(this._logger, id.toString()); - const res = await callWithMetadata( + const res = await this._call( this._stub.getEntity.bind(this._stub), req, this._metadataGenerator, @@ -1271,7 +1316,7 @@ export class TaskHubGrpcClient { req.setQuery(protoQuery); - const res = await callWithMetadata( + const res = await this._call( this._stub.queryEntities.bind(this._stub), req, this._metadataGenerator, @@ -1311,7 +1356,7 @@ export class TaskHubGrpcClient { protoReq.setContinuationtoken(token); } - const res = await callWithMetadata( + const res = await this._call( this._stub.cleanEntityStorage.bind(this._stub), protoReq, this._metadataGenerator, diff --git a/packages/durabletask-js/src/payloads/history-stream.ts b/packages/durabletask-js/src/payloads/history-stream.ts new file mode 100644 index 0000000..81d09ea --- /dev/null +++ b/packages/durabletask-js/src/payloads/history-stream.ts @@ -0,0 +1,62 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import * as grpc from "@grpc/grpc-js"; +import * as pb from "../proto/orchestrator_service_pb"; +import { PayloadProcessor } from "./payload-processor"; + +export class PayloadHydrationError extends Error { + constructor(cause: unknown) { + super("History payload hydration failed.", { cause }); + this.name = "PayloadHydrationError"; + } +} + +/** Serializes chunk hydration, applies stream backpressure, and drains I/O before settling. */ +export function readPayloadHistory( + stream: grpc.ClientReadableStream, + processor: PayloadProcessor, + signal?: AbortSignal, +): Promise { + return new Promise((resolve, reject) => { + const events: pb.HistoryEvent[] = []; + let pending = Promise.resolve(); + let finishing = false; + const finish = (error?: unknown) => { + if (finishing) return; + finishing = true; + stream.removeListener("data", onData); + stream.removeListener("end", onEnd); + stream.removeListener("error", onError); + stream.removeListener("close", onClose); + stream.on("error", () => {}); + signal?.removeEventListener("abort", onAbort); + if (error !== undefined) stream.cancel(); + stream.destroy(); + void pending.then(() => (error === undefined ? resolve(events) : reject(error)), reject); + }; + const onData = (chunk: pb.HistoryChunk) => { + stream.pause(); + pending = pending.then(async () => { + try { + await processor.resolve(chunk, signal); + } catch (error) { + throw new PayloadHydrationError(error); + } + for (const event of chunk.getEventsList()) events.push(event); + if (!finishing) stream.resume(); + }); + void pending.catch(finish); + }; + const onEnd = () => finish(); + const onError = (error: unknown) => finish(error); + const onClose = () => finish(new Error("History stream closed before all history was received.")); + const onAbort = () => finish(signal?.reason ?? new Error("History hydration was cancelled.")); + stream.on("data", onData); + stream.once("end", onEnd); + stream.once("error", onError); + stream.once("close", onClose); + signal?.addEventListener("abort", onAbort, { once: true }); + if (signal?.aborted) onAbort(); + }); +} 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 962b6fe..46284b6 100644 --- a/packages/durabletask-js/src/worker/task-hub-grpc-worker.ts +++ b/packages/durabletask-js/src/worker/task-hub-grpc-worker.ts @@ -26,6 +26,9 @@ import { ExponentialBackoff, sleep, withTimeout } from "../utils/backoff.util"; import { VersioningOptions, VersionMatchStrategy, VersionFailureStrategy } from "./versioning-options"; import { WorkItemFilters, generateWorkItemFiltersFromRegistry, toGrpcWorkItemFilters } from "./work-item-filters"; import { compareVersions } from "../utils/versioning.util"; +import { PayloadStorageOptions } from "../payloads/payload-store"; +import { PayloadProcessor } from "../payloads/payload-processor"; +import { readPayloadHistory, PayloadHydrationError } from "../payloads/history-stream"; import * as WorkerLogs from "./logs"; import { ConcurrencyOptions, ResolvedConcurrencyOptions, resolveConcurrencyOptions } from "./concurrency-options"; import { @@ -61,6 +64,8 @@ type WorkItemStreamResult = { * Options for creating a TaskHubGrpcWorker. */ export interface TaskHubGrpcWorkerOptions { + /** Optional externalized payload storage. Disabled by default. */ + payloads?: PayloadStorageOptions; /** * Maximum backend timer segment in milliseconds. Defaults to three days, like Python. * null, zero, or negative values disable segmentation. Must be finite; positive @@ -139,6 +144,7 @@ export class TaskHubGrpcWorker { private _workerLoopPromise: Promise | null; private _deferredStubCloseTimers: Map>; private readonly _maximumTimerIntervalMs: number | null; + private readonly _payloads?: PayloadProcessor; /** * Creates a new TaskHubGrpcWorker instance. @@ -207,6 +213,7 @@ export class TaskHubGrpcWorker { resolvedWorkItemFilters = hostAddressOrOptions.workItemFilters; resolvedConcurrency = hostAddressOrOptions.concurrency; resolvedMaximumTimerIntervalMs = hostAddressOrOptions.maximumTimerIntervalMs; + if (hostAddressOrOptions.payloads) this._payloads = new PayloadProcessor(hostAddressOrOptions.payloads); } else { // Deprecated positional parameters constructor resolvedHostAddress = hostAddressOrOptions; @@ -857,6 +864,7 @@ export class TaskHubGrpcWorker { private _buildGetWorkItemsRequest(): pb.GetWorkItemsRequest { const request = new pb.GetWorkItemsRequest(); request.setCapabilitiesList([pb.WorkerCapability.WORKER_CAPABILITY_HISTORY_STREAMING]); + if (this._payloads) request.addCapabilities(pb.WorkerCapability.WORKER_CAPABILITY_LARGE_PAYLOADS); request.setMaxconcurrentactivityworkitems( Math.min(this._concurrency.maximumConcurrentActivityWorkItems, MAX_PROTOCOL_CONCURRENCY), ); @@ -974,6 +982,7 @@ export class TaskHubGrpcWorker { request: TReq, signal?: AbortSignal, ): Promise { + if (this._payloads) await this._payloads.externalize(request, signal); const backoff = new ExponentialBackoff({ initialDelayMs: 200, maxDelayMs: 15000, @@ -1059,6 +1068,13 @@ export class TaskHubGrpcWorker { if (settled) return; signal?.throwIfAborted(); stream = stub.streamInstanceHistory(request, metadata); + if (this._payloads) { + readPayloadHistory(stream, this._payloads, signal).then((hydrated) => { + for (const event of hydrated) events.push(event); + finish(); + }, onError); + return; + } stream.on("data", onData); stream.once("end", onEnd); stream.once("error", onError); @@ -1079,6 +1095,8 @@ export class TaskHubGrpcWorker { ): Promise { const instanceId = req.getInstanceid(); + if (this._payloads) await this._payloads.resolve(req, retrySignal); + if (!instanceId) { throw new Error(`Could not execute the orchestrator as the instanceId was not provided (${instanceId})`); } @@ -1096,6 +1114,7 @@ export class TaskHubGrpcWorker { req.setPasteventsList(pastEvents); } catch (e: unknown) { if (retrySignal?.aborted) return; + if (e instanceof PayloadHydrationError) throw e.cause; const error = e instanceof Error ? e : new Error(String(e)); WorkerLogs.executionError(this._logger, instanceId, error); const res = new pb.OrchestratorResponse(); @@ -1327,6 +1346,7 @@ export class TaskHubGrpcWorker { stub: stubs.TaskHubSidecarServiceClient, retrySignal?: AbortSignal, ): Promise { + if (this._payloads) await this._payloads.resolve(req, retrySignal); const instanceId = req.getOrchestrationinstance()?.getInstanceid(); if (!instanceId) { @@ -1335,12 +1355,7 @@ export class TaskHubGrpcWorker { const versionCheck = this._checkVersionCompatibility(req.getVersion()?.getValue()); if (!versionCheck.compatible && !versionCheck.shouldFail) { - WorkerLogs.versionMismatchAbandon( - this._logger, - instanceId, - versionCheck.errorType!, - versionCheck.errorMessage!, - ); + WorkerLogs.versionMismatchAbandon(this._logger, instanceId, versionCheck.errorType!, versionCheck.errorMessage!); try { const abandon = new pb.AbandonActivityTaskRequest().setCompletiontoken(completionToken); await this._deliverResponse(stub.abandonTaskActivityWorkItem.bind(stub), abandon, retrySignal); @@ -1444,6 +1459,7 @@ export class TaskHubGrpcWorker { operationInfos?: pb.OperationInfo[], retrySignal?: AbortSignal, ): Promise { + if (this._payloads) await this._payloads.resolve(req, retrySignal); const instanceIdString = req.getInstanceid(); if (!instanceIdString) { @@ -1546,6 +1562,7 @@ export class TaskHubGrpcWorker { stub: stubs.TaskHubSidecarServiceClient, retrySignal?: AbortSignal, ): Promise { + if (this._payloads) await this._payloads.resolve(req, retrySignal); // Convert EntityRequest (V2) to EntityBatchRequest (V1) format const batchRequest = new pb.EntityBatchRequest(); batchRequest.setInstanceid(req.getInstanceid()); diff --git a/packages/durabletask-js/test/payload-transport.spec.ts b/packages/durabletask-js/test/payload-transport.spec.ts new file mode 100644 index 0000000..a1bc197 --- /dev/null +++ b/packages/durabletask-js/test/payload-transport.spec.ts @@ -0,0 +1,233 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import * as grpc from "@grpc/grpc-js"; +import { Empty } from "google-protobuf/google/protobuf/empty_pb"; +import { StringValue } from "google-protobuf/google/protobuf/wrappers_pb"; +import { TaskHubGrpcClient, TaskHubGrpcWorker, PayloadStore, NoOpLogger } from "../src"; +import * as pb from "../src/proto/orchestrator_service_pb"; +import * as stubs from "../src/proto/orchestrator_service_grpc_pb"; +import * as pbh from "../src/utils/pb-helper.util"; + +const text = (value: string) => new StringValue().setValue(value); +class MemoryStore extends PayloadStore { + readonly values = new Map(); + async upload(value: string): Promise { + const token = `payload:${this.values.size}`; + this.values.set(token, value); + return token; + } + async download(token: string): Promise { + await new Promise((resolve) => setTimeout(resolve, 5)); + if (!this.values.has(token)) throw new Error("Missing blob"); + return this.values.get(token)!; + } + isKnownPayloadToken(value: string): boolean { + return value.startsWith("payload:"); + } +} + +async function waitFor(predicate: () => boolean): Promise { + const deadline = Date.now() + 3000; + while (!predicate()) { + if (Date.now() > deadline) throw new Error("Timed out"); + await new Promise((resolve) => setTimeout(resolve, 5)); + } +} + +describe("transparent payload transport over gRPC", () => { + let server: grpc.Server; + let client: TaskHubGrpcClient; + let worker: TaskHubGrpcWorker; + let store: MemoryStore; + let hostAddress: string; + let input: string; + let subscription: grpc.ServerWritableStream | undefined; + let activityResponse: pb.ActivityResponse | undefined; + let activityCalls: number; + let requestedInput: string | undefined; + let orchestrationResponses: pb.OrchestratorResponse[]; + let historyEvents: pb.HistoryEvent[]; + + beforeEach(async () => { + store = new MemoryStore(); + input = await store.upload(JSON.stringify({ message: "hydrated" })); + subscription = undefined; + activityResponse = undefined; + activityCalls = 0; + requestedInput = undefined; + orchestrationResponses = []; + historyEvents = [pbh.newExecutionStartedEvent("flow", "instance", input)]; + server = new grpc.Server(); + const state = () => + new pb.OrchestrationState() + .setInstanceid("instance") + .setName("workflow") + .setInput(text(input)) + .setOutput(text(input)) + .setCustomstatus(text(input)); + const service = { + hello: (_call, callback) => callback(null, new Empty()), + startInstance: (call, callback) => { + requestedInput = call.request.getInput()?.getValue(); + callback(null, new pb.CreateInstanceResponse().setInstanceid(call.request.getInstanceid())); + }, + getInstance: (_call, callback) => + callback(null, new pb.GetInstanceResponse().setExists(true).setOrchestrationstate(state())), + waitForInstanceStart: (_call, callback) => + callback(null, new pb.GetInstanceResponse().setExists(true).setOrchestrationstate(state())), + queryInstances: (_call, callback) => + callback(null, new pb.QueryInstancesResponse().setOrchestrationstateList([state()])), + streamInstanceHistory: (call) => { + call.write(new pb.HistoryChunk().setEventsList(historyEvents)); + call.end(); + }, + getWorkItems: (call) => { + subscription = call; + call.on("cancelled", () => call.end()); + }, + completeActivityTask: (call, callback) => { + activityResponse = call.request; + callback(null, new pb.CompleteTaskResponse()); + }, + completeOrchestratorTask: (call, callback) => { + orchestrationResponses.push(call.request); + callback(null, new pb.CompleteTaskResponse()); + }, + } satisfies Pick< + stubs.ITaskHubSidecarServiceServer, + | "hello" + | "startInstance" + | "getInstance" + | "waitForInstanceStart" + | "queryInstances" + | "streamInstanceHistory" + | "getWorkItems" + | "completeActivityTask" + | "completeOrchestratorTask" + >; + server.addService(stubs.TaskHubSidecarServiceService, service); + const port = await new Promise((resolve, reject) => + server.bindAsync("127.0.0.1:0", grpc.ServerCredentials.createInsecure(), (error, port) => + error ? reject(error) : resolve(port), + ), + ); + hostAddress = `127.0.0.1:${port}`; + const payloads = { store, thresholdBytes: 1 }; + client = new TaskHubGrpcClient({ hostAddress, payloads, logger: new NoOpLogger() }); + worker = new TaskHubGrpcWorker({ hostAddress, payloads, logger: new NoOpLogger() }); + worker.addNamedActivity("echo", async (_context, value) => { + activityCalls++; + return value; + }); + }); + + afterEach(async () => { + if (worker["_isRunning"]) await worker.stop(); + await client.stop(); + server.forceShutdown(); + }); + + it("externalizes before sending and hydrates state, wait, query and streamed history before conversion", async () => { + await client.scheduleNewOrchestration("workflow", { message: "request" }); + expect(requestedInput).toMatch(/^payload:/); + expect(await store.download(requestedInput!)).toBe('{"message":"request"}'); + expect((await client.getOrchestrationState("instance"))!.serializedInput).toBe('{"message":"hydrated"}'); + expect((await client.waitForOrchestrationStart("instance"))!.serializedInput).toBe('{"message":"hydrated"}'); + const states = []; + for await (const item of client.getAllInstances({ fetchInputsAndOutputs: true })) states.push(item); + expect(states[0].serializedInput).toBe('{"message":"hydrated"}'); + const history = await client.getOrchestrationHistory("instance"); + expect(history[0]).toMatchObject({ input: '{"message":"hydrated"}' }); + }); + + it("advertises support, awaits input hydration and externalizes worker output", async () => { + await worker.start(); + await waitFor(() => !!subscription); + expect(subscription!.request.getCapabilitiesList()).toContain(pb.WorkerCapability.WORKER_CAPABILITY_LARGE_PAYLOADS); + subscription!.write( + new pb.WorkItem() + .setCompletiontoken("completion") + .setActivityrequest( + new pb.ActivityRequest() + .setName("echo") + .setTaskid(1) + .setInput(text(input)) + .setOrchestrationinstance(new pb.OrchestrationInstance().setInstanceid("instance")), + ), + ); + await waitFor(() => !!activityResponse); + expect(activityCalls).toBe(1); + expect(activityResponse!.getCompletiontoken()).toBe("completion"); + expect(await store.download(activityResponse!.getResult()!.getValue())).toBe('{"message":"hydrated"}'); + }); + + it("propagates missing streamed payloads rather than returning tokens or succeeding early at end", async () => { + store.values.clear(); + await expect(client.getOrchestrationHistory("instance")).rejects.toThrow("Missing blob"); + }); + + it("hydrates streamed orchestration history before user execution and externalizes its result", async () => { + let observed: unknown; + worker.addNamedOrchestrator("flow", async (_ctx, value) => { + observed = value; + return value; + }); + await worker.start(); + await waitFor(() => !!subscription); + subscription!.write( + new pb.WorkItem() + .setCompletiontoken("streamed") + .setOrchestratorrequest( + new pb.OrchestratorRequest() + .setInstanceid("instance") + .setRequireshistorystreaming(true) + .setNeweventsList([pbh.newOrchestratorStartedEvent()]), + ), + ); + await waitFor(() => orchestrationResponses.length > 0); + expect(observed).toEqual({ message: "hydrated" }); + const result = orchestrationResponses[0].getActionsList()[0].getCompleteorchestration()!.getResult()!.getValue(); + expect(await store.download(result)).toBe('{"message":"hydrated"}'); + }); + + it("leaves a streamed payload download failure retryable instead of failing the orchestration", async () => { + const errors = jest.fn(); + worker = new TaskHubGrpcWorker({ + hostAddress, + payloads: { store }, + logger: { + error: errors, + warn: () => {}, + info: () => {}, + debug: () => {}, + }, + }); + const flow = jest.fn(async () => "should not run"); + worker.addNamedOrchestrator("flow", flow); + store.values.clear(); + await worker.start(); + await waitFor(() => !!subscription); + subscription!.write( + new pb.WorkItem() + .setCompletiontoken("retryable") + .setOrchestratorrequest( + new pb.OrchestratorRequest() + .setInstanceid("instance") + .setRequireshistorystreaming(true) + .setNeweventsList([pbh.newOrchestratorStartedEvent()]), + ), + ); + await waitFor(() => errors.mock.calls.length > 0); + await waitFor(() => worker["_pendingWorkItems"].size === 0); + expect(flow).not.toHaveBeenCalled(); + expect(orchestrationResponses).toEqual([]); + }); + + it("collects large history chunks without the JavaScript variadic argument limit", async () => { + historyEvents = Array.from({ length: 150000 }, () => + new pb.HistoryEvent().setGenericevent(new pb.GenericEvent().setData(text("x"))), + ); + expect(await client.getOrchestrationHistory("instance")).toHaveLength(historyEvents.length); + }, 20000); +});