diff --git a/CHANGELOG.md b/CHANGELOG.md index 356f4162..52f244be 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -47,6 +47,16 @@ ### Fixes +- Share the unary client payload/metadata/RPC/hydration pipeline across scheduling, waits, rewind, + and restart. Only actual RPC failures enter operation-specific error translation or retry; + storage and metadata failures keep their original identity, and cancellation prevents retries. +- Restrict scheduling error translation to `startInstance` RPC failures. Payload upload and + metadata errors retain their original identity instead of becoming duplicate-ID or argument errors. +- Limit completion-wait deadline retries to scheduler RPC failures. Payload download errors, + including custom-store gRPC deadlines, propagate unchanged after the successful wait response. +- Abandon acquired worker work items when enabled payload hydration or unconverted output storage + processing fails, awaiting shutdown-aware bounded release retries without creating fake business + failures. Tokenless host processing preserves unconverted storage errors. - Apply payload externalization and UTF-8 size limits to native continue-as-new carryover history. Event metadata, ordering, and hydrated payload values are preserved. - Validate every eligible payload's UTF-8 size before starting uploads, preventing storage writes diff --git a/README.md b/README.md index 823f6dea..2759742a 100644 --- a/README.md +++ b/README.md @@ -23,6 +23,31 @@ 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. + +Workers abandon acquired queue work items when payload downloads or unconverted output storage/size +checks fail. Release is awaited with bounded gRPC retries and the original work item's cancellation +signal; release failures are logged, and shutdown prevents late sends. Permanent activity and +orchestration output failures still produce non-retriable failed completions. Tokenless host APIs +(`processOrchestratorRequest` and `processEntityBatchRequest`) instead reject with the original +unconverted storage error so the host can apply its own retry policy. + +All unary client operations share the same payload and metadata pipeline. Only actual scheduler +RPC failures enter operation-specific error translation or retry; storage and metadata errors +propagate unchanged. Completion waits retry scheduler deadlines with fresh metadata, without +re-uploading the request or retrying response hydration. Cancellation stops RPCs and retry backoff. + +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 e1d04436..29223192 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 60244b3c..a57bd6cb 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 29f72a6e..151e9a49 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 eab641be..1c848513 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 8a0b1a76..05e61418 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 00000000..d7975126 --- /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 54d3bb7c..0b56ad04 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,43 @@ export class TaskHubGrpcClient { await new Promise((resolve) => setTimeout(resolve, 1000)); } + // The RPC-only handler may throw a mapped error, return true to retry the RPC, + // or return false to rethrow the original error. + private async _call( + method: Parameters>[0], + request: TReq, + metadataGenerator?: MetadataGenerator, + signal?: AbortSignal, + handleRpcError?: (error: unknown, effectiveSignal?: AbortSignal) => boolean | Promise, + ): Promise { + const effectiveSignal = this._payloads + ? signal + ? AbortSignal.any([signal, this._payloadCancellation.signal]) + : this._payloadCancellation.signal + : signal; + effectiveSignal?.throwIfAborted(); + if (this._payloads) await this._payloads.externalize(request, effectiveSignal); + while (true) { + effectiveSignal?.throwIfAborted(); + let attemptMetadata: MetadataGenerator | undefined; + if (metadataGenerator) { + const metadata = await metadataGenerator(); + attemptMetadata = async () => metadata; + } + let response: TRes; + try { + response = await callWithMetadata(method, request, attemptMetadata, effectiveSignal); + } catch (error) { + effectiveSignal?.throwIfAborted(); + if ((await handleRpcError?.(error, effectiveSignal)) === true) continue; + throw error; + } + effectiveSignal?.throwIfAborted(); + if (this._payloads) await this._payloads.resolve(response, effectiveSignal); + return response; + } + } + /** * Schedules a new orchestrator using the DurableTask client. * @@ -256,27 +302,31 @@ 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, + undefined, + (e: unknown) => { + if (e instanceof Error && "code" in e) { + const grpcError = e as grpc.ServiceError; + const message = grpcError.details || grpcError.message; + if (grpcError.code === grpc.status.ALREADY_EXISTS) { + throw new OrchestrationAlreadyExistsError(message, { cause: e }); + } + if (grpcError.code === grpc.status.INVALID_ARGUMENT) { + throw new TypeError(message, { cause: e }); + } + // The JS SDK has no named cancellation error contract, so preserve CANCELLED as a ServiceError. + } + return false; + }, ); setSpanOk(span); return res.getInstanceid(); } catch (e: unknown) { setSpanError(span, e); - if (e instanceof Error && "code" in e) { - const grpcError = e as grpc.ServiceError; - const message = grpcError.details || grpcError.message; - if (grpcError.code === grpc.status.ALREADY_EXISTS) { - throw new OrchestrationAlreadyExistsError(message, { cause: e }); - } - if (grpcError.code === grpc.status.INVALID_ARGUMENT) { - throw new TypeError(message, { cause: e }); - } - // The JS SDK has no named cancellation error contract, so preserve CANCELLED as a ServiceError. - } throw e; } finally { endSpan(span); @@ -301,7 +351,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 +393,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); @@ -359,6 +409,7 @@ export class TaskHubGrpcClient { * * If an orchestrator instance is already in a terminal state when this method is called, it returns immediately. * Server deadline errors are retried with backoff within the original total timeout. + * Payload storage errors propagate without retrying the successful wait RPC. * * @param {string} instanceId - The unique identifier of the orchestrator instance to wait for. * @param {boolean} fetchPayloads - Indicates whether to fetch the orchestrator instance's @@ -410,26 +461,19 @@ export class TaskHubGrpcClient { private async _waitForCompletion(req: pb.GetInstanceRequest, signal: AbortSignal): Promise { const backoff = new ExponentialBackoff({ initialDelayMs: 100, maxDelayMs: 1000, jitterFactor: 0 }); - while (true) { - signal.throwIfAborted(); - // Metadata failures must not enter the remote-deadline retry path. - const metadata = this._metadataGenerator ? await this._metadataGenerator() : new grpc.Metadata(); - try { - return await callWithMetadata( - this._stub.waitForInstanceCompletion.bind(this._stub), - req, - async () => metadata, - signal, - ); - } catch (error) { - signal.throwIfAborted(); + return this._call( + this._stub.waitForInstanceCompletion.bind(this._stub), + req, + this._metadataGenerator, + signal, + async (error, effectiveSignal) => { if (error instanceof Error && "code" in error && error.code === grpc.status.DEADLINE_EXCEEDED) { - await backoff.wait(signal); - continue; + await backoff.wait(effectiveSignal); + return true; } - throw error; - } - } + return false; + }, + ); } private async _withWaitCancellation( @@ -498,7 +542,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 +609,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 +635,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 +661,7 @@ export class TaskHubGrpcClient { ClientLogs.resumingInstance(this._logger, instanceId); - await callWithMetadata( + await this._call( this._stub.resumeInstance.bind(this._stub), req, this._metadataGenerator, @@ -655,35 +699,34 @@ export class TaskHubGrpcClient { ClientLogs.rewindingInstance(this._logger, instanceId, reason); - try { - await callWithMetadata( - this._stub.rewindInstance.bind(this._stub), - req, - this._metadataGenerator, - ); - } catch (e) { - // Handle gRPC errors and convert them to appropriate errors - if (e instanceof Error && "code" in e) { - const grpcError = e as grpc.ServiceError; - if (grpcError.code === grpc.status.NOT_FOUND) { - throw new Error(`An orchestration with the instanceId '${instanceId}' was not found.`, { cause: e }); - } - if (grpcError.code === grpc.status.FAILED_PRECONDITION) { - throw new Error( - grpcError.details || - `Cannot rewind orchestration '${instanceId}': it is in a state that does not allow rewinding.`, - { cause: e }, - ); - } - if (grpcError.code === grpc.status.UNIMPLEMENTED) { - throw new Error(grpcError.details || `The rewind operation is not supported by the backend.`, { cause: e }); - } - if (grpcError.code === grpc.status.CANCELLED) { - throw new Error(`The rewind operation for '${instanceId}' was canceled.`, { cause: e }); + await this._call( + this._stub.rewindInstance.bind(this._stub), + req, + this._metadataGenerator, + undefined, + (e: unknown) => { + if (e instanceof Error && "code" in e) { + const grpcError = e as grpc.ServiceError; + if (grpcError.code === grpc.status.NOT_FOUND) { + throw new Error(`An orchestration with the instanceId '${instanceId}' was not found.`, { cause: e }); + } + if (grpcError.code === grpc.status.FAILED_PRECONDITION) { + throw new Error( + grpcError.details || + `Cannot rewind orchestration '${instanceId}': it is in a state that does not allow rewinding.`, + { cause: e }, + ); + } + if (grpcError.code === grpc.status.UNIMPLEMENTED) { + throw new Error(grpcError.details || `The rewind operation is not supported by the backend.`, { cause: e }); + } + if (grpcError.code === grpc.status.CANCELLED) { + throw new Error(`The rewind operation for '${instanceId}' was canceled.`, { cause: e }); + } } - } - throw e; - } + return false; + }, + ); } /** @@ -720,28 +763,28 @@ export class TaskHubGrpcClient { ClientLogs.restartingInstance(this._logger, instanceId, restartWithNewInstanceId); - try { - const res = await callWithMetadata( - this._stub.restartInstance.bind(this._stub), - req, - this._metadataGenerator, - ); - return res.getInstanceid(); - } catch (e) { - if (e instanceof Error && "code" in e) { - const grpcError = e as grpc.ServiceError; - if (grpcError.code === grpc.status.NOT_FOUND) { - throw new Error(`An orchestration with the instanceId '${instanceId}' was not found.`, { cause: e }); - } - if (grpcError.code === grpc.status.FAILED_PRECONDITION) { - throw new Error(`An orchestration with the instanceId '${instanceId}' cannot be restarted.`, { cause: e }); - } - if (grpcError.code === grpc.status.CANCELLED) { - throw new Error(`The restartOrchestration operation was canceled.`, { cause: e }); + const res = await this._call( + this._stub.restartInstance.bind(this._stub), + req, + this._metadataGenerator, + undefined, + (e: unknown) => { + if (e instanceof Error && "code" in e) { + const grpcError = e as grpc.ServiceError; + if (grpcError.code === grpc.status.NOT_FOUND) { + throw new Error(`An orchestration with the instanceId '${instanceId}' was not found.`, { cause: e }); + } + if (grpcError.code === grpc.status.FAILED_PRECONDITION) { + throw new Error(`An orchestration with the instanceId '${instanceId}' cannot be restarted.`, { cause: e }); + } + if (grpcError.code === grpc.status.CANCELLED) { + throw new Error(`The restartOrchestration operation was canceled.`, { cause: e }); + } } - } - throw e; - } + return false; + }, + ); + return res.getInstanceid(); } /** @@ -775,7 +818,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 +854,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 +958,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 +1044,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 +1092,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 +1210,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 +1246,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 +1334,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 +1374,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 00000000..81d09eaa --- /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 962b6fec..b5b8d35b 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), ); @@ -956,11 +964,23 @@ export class TaskHubGrpcWorker { return undefined; } - private _trackPendingWorkItem(workPromise: Promise, onError: (error: Error) => void): void { + private _trackPendingWorkItem( + workPromise: Promise, + onError: (error: Error) => void, + abandon: () => Promise, + signal?: AbortSignal, + ): void { const handledPromise = workPromise - .catch((e: unknown) => { + .catch(async (e: unknown) => { const error = e instanceof Error ? e : new Error(String(e)); onError(error); + if (this._payloads && !signal?.aborted) { + try { + await abandon(); + } catch (error: unknown) { + WorkerLogs.workerError(this._logger, error); + } + } }) .finally(() => { this._pendingWorkItems.delete(handledPromise); @@ -1010,9 +1030,17 @@ export class TaskHubGrpcWorker { ): void { const retrySignal = this._abortController?.signal; const workPromise = this._executeOrchestratorInternal(req, completionToken, stub, retrySignal); - this._trackPendingWorkItem(workPromise, (error) => { - WorkerLogs.executionError(this._logger, req.getInstanceid() || "(unknown)", error); - }); + this._trackPendingWorkItem( + workPromise, + (error) => WorkerLogs.executionError(this._logger, req.getInstanceid() || "(unknown)", error), + () => + this._deliverResponse( + stub.abandonTaskOrchestratorWorkItem.bind(stub), + new pb.AbandonOrchestrationTaskRequest().setCompletiontoken(completionToken), + retrySignal, + ), + retrySignal, + ); } private async _streamOrchestrationHistory( @@ -1059,6 +1087,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 +1114,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 +1133,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(); @@ -1295,6 +1333,8 @@ export class TaskHubGrpcWorker { endSpan(tracingResult?.span); } + // Storage failures must reach the queue boundary, not become failed business completions. + if (this._payloads) await this._payloads.externalize(res, retrySignal); try { await this._deliverResponse(stub.completeOrchestratorTask.bind(stub), res, retrySignal); } catch (e: unknown) { @@ -1313,9 +1353,17 @@ export class TaskHubGrpcWorker { ): void { const retrySignal = this._abortController?.signal; const workPromise = this._executeActivityInternal(req, completionToken, stub, retrySignal); - this._trackPendingWorkItem(workPromise, (error) => { - WorkerLogs.workerError(this._logger, error); - }); + this._trackPendingWorkItem( + workPromise, + (error) => WorkerLogs.workerError(this._logger, error), + () => + this._deliverResponse( + stub.abandonTaskActivityWorkItem.bind(stub), + new pb.AbandonActivityTaskRequest().setCompletiontoken(completionToken), + retrySignal, + ), + retrySignal, + ); } /** @@ -1327,6 +1375,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 +1384,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); @@ -1401,6 +1445,7 @@ export class TaskHubGrpcWorker { endSpan(activitySpan); } + if (this._payloads) await this._payloads.externalize(res, retrySignal); try { await this._deliverResponse(stub.completeActivityTask.bind(stub), res, retrySignal); } catch (e: unknown) { @@ -1420,9 +1465,17 @@ export class TaskHubGrpcWorker { ): void { const retrySignal = this._abortController?.signal; const workPromise = this._executeEntityInternal(req, completionToken, stub, operationInfos, retrySignal); - this._trackPendingWorkItem(workPromise, (error) => { - WorkerLogs.workerError(this._logger, error); - }); + this._trackPendingWorkItem( + workPromise, + (error) => WorkerLogs.workerError(this._logger, error), + () => + this._deliverResponse( + stub.abandonTaskEntityWorkItem.bind(stub), + new pb.AbandonEntityTaskRequest().setCompletiontoken(completionToken), + retrySignal, + ), + retrySignal, + ); } /** @@ -1444,6 +1497,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) { @@ -1523,9 +1577,17 @@ export class TaskHubGrpcWorker { ): void { const retrySignal = this._abortController?.signal; const workPromise = this._executeEntityV2Internal(req, completionToken, stub, retrySignal); - this._trackPendingWorkItem(workPromise, (error) => { - WorkerLogs.workerError(this._logger, error); - }); + this._trackPendingWorkItem( + workPromise, + (error) => WorkerLogs.workerError(this._logger, error), + () => + this._deliverResponse( + stub.abandonTaskEntityWorkItem.bind(stub), + new pb.AbandonEntityTaskRequest().setCompletiontoken(completionToken), + retrySignal, + ), + retrySignal, + ); } /** @@ -1546,6 +1608,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()); @@ -1680,6 +1743,7 @@ export class TaskHubGrpcWorker { stub: stubs.TaskHubSidecarServiceClient, retrySignal?: AbortSignal, ): Promise { + if (this._payloads) await this._payloads.externalize(batchResult, retrySignal); try { await this._deliverResponse(stub.completeEntityTask.bind(stub), batchResult, retrySignal); } catch (e: any) { diff --git a/packages/durabletask-js/test/client-error-cause.spec.ts b/packages/durabletask-js/test/client-error-cause.spec.ts index 7be54bbb..c5d4942f 100644 --- a/packages/durabletask-js/test/client-error-cause.spec.ts +++ b/packages/durabletask-js/test/client-error-cause.spec.ts @@ -2,7 +2,10 @@ // Licensed under the MIT License. import * as grpc from "@grpc/grpc-js"; -import { TaskHubGrpcClient } from "../src"; +import { EventEmitter } from "events"; +import { StringValue } from "google-protobuf/google/protobuf/wrappers_pb"; +import { NoOpLogger, PayloadStore, TaskHubGrpcClient } from "../src"; +import * as pb from "../src/proto/orchestrator_service_pb"; /** * Creates a mock gRPC ServiceError with the specified status code and details. @@ -22,6 +25,232 @@ function getStub(client: TaskHubGrpcClient): any { return (client as any)._stub; } +class ErrorDomainPayloadStore extends PayloadStore { + readonly values = new Map(); + upload = jest.fn(async (value: string, _signal?: AbortSignal) => { + const token = `payload:${this.values.size}`; + this.values.set(token, value); + return token; + }); + download = jest.fn(async (token: string, _signal?: AbortSignal) => { + if (!this.values.has(token)) throw new Error(`Missing payload: ${token}`); + return this.values.get(token)!; + }); + isKnownPayloadToken(value: string): boolean { + return value.startsWith("payload:"); + } +} + +type ManagementRequest = pb.RewindInstanceRequest | pb.RestartInstanceRequest; +type ManagementResponse = pb.RewindInstanceResponse | pb.RestartInstanceResponse; +type ManagementCallback = (error: grpc.ServiceError | null, response: ManagementResponse) => void; + +const managementOperations: { + name: "rewind" | "restart"; + rpc: "rewindInstance" | "restartInstance"; + invoke: (client: TaskHubGrpcClient) => Promise; + response: () => ManagementResponse; + mappings: { code: grpc.status; message: string }[]; +}[] = [ + { + name: "rewind", + rpc: "rewindInstance", + invoke: (client: TaskHubGrpcClient) => client.rewindInstance("test-instance", "test reason"), + response: () => new pb.RewindInstanceResponse(), + mappings: [ + { code: grpc.status.NOT_FOUND, message: "An orchestration with the instanceId 'test-instance' was not found." }, + { code: grpc.status.FAILED_PRECONDITION, message: "backend details" }, + { code: grpc.status.UNIMPLEMENTED, message: "backend details" }, + { code: grpc.status.CANCELLED, message: "The rewind operation for 'test-instance' was canceled." }, + ], + }, + { + name: "restart", + rpc: "restartInstance", + invoke: (client: TaskHubGrpcClient) => client.restartOrchestration("test-instance", true), + response: () => new pb.RestartInstanceResponse().setInstanceid("restarted-instance"), + mappings: [ + { code: grpc.status.NOT_FOUND, message: "An orchestration with the instanceId 'test-instance' was not found." }, + { + code: grpc.status.FAILED_PRECONDITION, + message: "An orchestration with the instanceId 'test-instance' cannot be restarted.", + }, + { code: grpc.status.CANCELLED, message: "The restartOrchestration operation was canceled." }, + ], + }, +]; + +describe.each([true, false])("unary management error domains with payloads enabled=%s", (payloadsEnabled) => { + let client: TaskHubGrpcClient; + let store: ErrorDomainPayloadStore; + let metadataGenerator: jest.Mock, []>; + const call: grpc.ClientUnaryCall = Object.assign(new EventEmitter(), { + cancel: jest.fn(), + getPeer: () => "test", + getAuthContext: () => null, + }); + + beforeEach(() => { + store = new ErrorDomainPayloadStore(); + metadataGenerator = jest.fn(async () => new grpc.Metadata()); + client = new TaskHubGrpcClient({ + payloads: payloadsEnabled ? { store, thresholdBytes: 1 } : undefined, + metadataGenerator, + logger: new NoOpLogger(), + }); + }); + + afterEach(() => { + client["_stub"].close(); + }); + + if (payloadsEnabled) { + it("externalizes eligible input once while retrying only RPCs with fresh metadata", async () => { + const request = new pb.CreateInstanceRequest() + .setName("workflow") + .setInstanceid("instance") + .setInput(new StringValue().setValue('"input"')); + const response = new pb.CreateInstanceResponse().setInstanceid("instance"); + const failure = Object.freeze(createGrpcError(grpc.status.DEADLINE_EXCEEDED)); + const sentTokens: (string | undefined)[] = []; + const rpc = jest.fn< + grpc.ClientUnaryCall, + [ + pb.CreateInstanceRequest, + grpc.Metadata, + (error: grpc.ServiceError | null, res: pb.CreateInstanceResponse) => void, + ] + >((req, _metadata, callback) => { + sentTokens.push(req.getInput()?.getValue()); + callback(sentTokens.length < 3 ? failure : null, response); + return call; + }); + metadataGenerator.mockImplementation(async () => { + const metadata = new grpc.Metadata(); + metadata.set("attempt", String(metadataGenerator.mock.calls.length)); + return metadata; + }); + const handleRpcError = jest.fn((error: unknown, signal?: AbortSignal) => { + expect(error).toBe(failure); + expect(signal).toBe(store.upload.mock.calls[0][1]); + return true; + }); + const resolve = jest.spyOn(client["_payloads"]!, "resolve"); + const controller = new AbortController(); + + await expect(client["_call"](rpc, request, metadataGenerator, controller.signal, handleRpcError)).resolves.toBe( + response, + ); + + expect(store.upload).toHaveBeenCalledTimes(1); + expect(store.values.get("payload:0")).toBe('"input"'); + expect(sentTokens).toEqual(["payload:0", "payload:0", "payload:0"]); + expect(rpc.mock.calls.every(([req]) => req === request)).toBe(true); + expect(rpc.mock.calls.map(([, metadata]) => metadata.get("attempt"))).toEqual([["1"], ["2"], ["3"]]); + expect(handleRpcError).toHaveBeenCalledTimes(2); + expect(resolve).toHaveBeenCalledTimes(1); + expect(resolve.mock.invocationCallOrder[0]).toBeGreaterThan(rpc.mock.invocationCallOrder[2]); + }); + } + + describe.each(managementOperations)("$name", ({ name, rpc: rpcName, invoke, response, mappings }) => { + let rpc: jest.Mock; + + beforeEach(() => { + rpc = jest.fn( + (_req, _metadata, callback) => { + callback(null, response()); + return call; + }, + ); + Object.defineProperty(client["_stub"], rpcName, { value: rpc }); + }); + + it.each(mappings)( + "preserves a reused frozen metadata error with mapped status $code without an RPC", + async ({ code }) => { + const failure = Object.freeze(createGrpcError(code, "metadata failed")); + metadataGenerator.mockRejectedValue(failure); + + await expect(invoke(client)).rejects.toBe(failure); + await expect(invoke(client)).rejects.toBe(failure); + expect(metadataGenerator).toHaveBeenCalledTimes(2); + expect(rpc).not.toHaveBeenCalled(); + expect(store.upload).toHaveBeenCalledTimes(payloadsEnabled && name === "rewind" ? 2 : 0); + expect(store.download).not.toHaveBeenCalled(); + }, + ); + + it.each(mappings)("maps genuine RPC status $code with the original frozen cause", async ({ code, message }) => { + const failure = Object.freeze(createGrpcError(code, "backend details")); + rpc.mockImplementation((_req, _metadata, callback) => { + callback(failure, response()); + return call; + }); + + const pending = invoke(client); + await expect(pending).rejects.toBeInstanceOf(Error); + await expect(pending).rejects.toMatchObject({ message, cause: failure }); + expect(rpc).toHaveBeenCalledTimes(1); + expect(metadataGenerator).toHaveBeenCalledTimes(1); + }); + + it("preserves an unmapped RPC error without retry or mutation", async () => { + const failure = Object.freeze(createGrpcError(grpc.status.INTERNAL, "unknown scheduler error")); + rpc.mockImplementation((_req, _metadata, callback) => { + callback(failure, response()); + return call; + }); + + await expect(invoke(client)).rejects.toBe(failure); + expect(rpc).toHaveBeenCalledTimes(1); + }); + + it("preserves successful request options and result", async () => { + const result = await invoke(client); + const request = rpc.mock.calls[0][0]; + expect(request.getInstanceid()).toBe("test-instance"); + if (request instanceof pb.RewindInstanceRequest) { + expect(request.getReason()?.getValue()).toBe(payloadsEnabled ? "payload:0" : "test reason"); + expect(store.upload).toHaveBeenCalledTimes(payloadsEnabled ? 1 : 0); + expect(result).toBeUndefined(); + } else { + expect(request.getRestartwithnewinstanceid()).toBe(true); + expect(store.upload).not.toHaveBeenCalled(); + expect(result).toBe("restarted-instance"); + } + expect(rpc).toHaveBeenCalledTimes(1); + }); + + if (payloadsEnabled) { + it.each(mappings)("does not translate a mapped-status $code abort reason from the RPC", async ({ code }) => { + const reason = Object.freeze(createGrpcError(code, "caller canceled")); + rpc.mockImplementation((_req, _metadata, callback) => { + client["_payloadCancellation"].abort(reason); + callback(createGrpcError(grpc.status.CANCELLED), response()); + return call; + }); + + await expect(invoke(client)).rejects.toBe(reason); + expect(rpc).toHaveBeenCalledTimes(1); + expect(metadataGenerator).toHaveBeenCalledTimes(1); + }); + } + + if (payloadsEnabled && name === "rewind") { + it.each(mappings)("preserves upload status $code without invoking metadata or the RPC", async ({ code }) => { + const failure = Object.freeze(createGrpcError(code, "upload failed")); + store.upload.mockRejectedValue(failure); + + await expect(invoke(client)).rejects.toBe(failure); + expect(store.upload).toHaveBeenCalledTimes(1); + expect(metadataGenerator).not.toHaveBeenCalled(); + expect(rpc).not.toHaveBeenCalled(); + }); + } + }); +}); + describe("TaskHubGrpcClient error cause preservation", () => { let client: TaskHubGrpcClient; @@ -51,10 +280,7 @@ describe("TaskHubGrpcClient error cause preservation", () => { }); it("should preserve error cause for FAILED_PRECONDITION gRPC errors", async () => { - const grpcError = createGrpcError( - grpc.status.FAILED_PRECONDITION, - "Orchestration is running", - ); + const grpcError = createGrpcError(grpc.status.FAILED_PRECONDITION, "Orchestration is running"); getStub(client).rewindInstance = (_req: any, _metadata: any, callback: any) => { callback(grpcError, null); @@ -150,10 +376,7 @@ describe("TaskHubGrpcClient error cause preservation", () => { }); it("should preserve error cause for FAILED_PRECONDITION gRPC errors", async () => { - const grpcError = createGrpcError( - grpc.status.FAILED_PRECONDITION, - "Orchestration is still running", - ); + const grpcError = createGrpcError(grpc.status.FAILED_PRECONDITION, "Orchestration is still running"); getStub(client).restartInstance = (_req: any, _metadata: any, callback: any) => { callback(grpcError, null); diff --git a/packages/durabletask-js/test/client-wait-grpc.spec.ts b/packages/durabletask-js/test/client-wait-grpc.spec.ts index 8d034199..0cf4d532 100644 --- a/packages/durabletask-js/test/client-wait-grpc.spec.ts +++ b/packages/durabletask-js/test/client-wait-grpc.spec.ts @@ -4,6 +4,7 @@ import * as grpc from "@grpc/grpc-js"; import { StringValue } from "google-protobuf/google/protobuf/wrappers_pb"; import { TaskHubGrpcClient } from "../src/client/client"; +import { PayloadStore } from "../src/payloads/payload-store"; import { TimeoutError } from "../src/exception/timeout-error"; import { NoOpLogger } from "../src/types/logger.type"; import * as pb from "../src/proto/orchestrator_service_pb"; @@ -11,6 +12,14 @@ import { TaskHubSidecarServiceService } from "../src/proto/orchestrator_service_ type WaitCall = grpc.ServerUnaryCall; +class WaitPayloadStore extends PayloadStore { + upload = jest.fn(async (_value: string, _signal?: AbortSignal) => "payload:output"); + download = jest.fn(async (_token: string, _signal?: AbortSignal) => '"hydrated"'); + isKnownPayloadToken(value: string): boolean { + return value.startsWith("payload:"); + } +} + function completedResponse(): pb.GetInstanceResponse { const state = new pb.OrchestrationState(); state.setName("grpc-wait"); @@ -25,6 +34,7 @@ function completedResponse(): pb.GetInstanceResponse { describe("Client waits over real gRPC", () => { const server = new grpc.Server(); let client: TaskHubGrpcClient; + let hostAddress: string; let handler: grpc.handleUnaryCall; let calls: WaitCall[]; let metadataCalls: number; @@ -46,8 +56,9 @@ describe("Client waits over real gRPC", () => { else resolve(boundPort); }); }); + hostAddress = `127.0.0.1:${port}`; client = new TaskHubGrpcClient({ - hostAddress: `127.0.0.1:${port}`, + hostAddress, options: { interceptors: [ (options: grpc.InterceptorOptions, nextCall: grpc.NextCall) => @@ -182,4 +193,104 @@ describe("Client waits over real gRPC", () => { }); expect(calls).toHaveLength(2); }); + + describe("externalized completion waits", () => { + let payloadClient: TaskHubGrpcClient; + let store: WaitPayloadStore; + let stopped: boolean; + + beforeEach(() => { + store = new WaitPayloadStore(); + stopped = false; + payloadClient = new TaskHubGrpcClient({ + hostAddress, + payloads: { store }, + metadataGenerator: client["_metadataGenerator"], + logger: new NoOpLogger(), + }); + handler = (_call, callback) => { + const response = completedResponse(); + response.getOrchestrationstate()!.setOutput(new StringValue().setValue("payload:output")); + callback(null, response); + }; + }); + + afterEach(async () => { + if (!stopped) await payloadClient.stop(); + }); + + it.each(["grpc deadline", "HTTP 503"] as const)( + "propagates the original storage %s after one successful wait RPC and download", + async (kind) => { + const failure = + kind === "grpc deadline" + ? Object.assign(new Error("storage deadline"), { + code: grpc.status.DEADLINE_EXCEEDED, + details: "storage deadline", + metadata: new grpc.Metadata(), + }) + : Object.assign(new Error("storage unavailable"), { statusCode: 503 }); + store.download.mockRejectedValue(failure); + await expect(payloadClient.waitForOrchestrationCompletion("grpc-instance", true, 0.45)).rejects.toBe(failure); + expect(calls).toHaveLength(1); + expect(store.download).toHaveBeenCalledTimes(1); + expect(metadataCalls).toBe(1); + }, + ); + + it("retries genuine wait RPC deadlines before hydrating the successful response once", async () => { + const success = handler; + handler = (call, callback) => { + if (calls.length < 3) callback({ code: grpc.status.DEADLINE_EXCEEDED, details: "remote deadline" }); + else success(call, callback); + }; + await expect(payloadClient.waitForOrchestrationCompletion("grpc-instance", true, 5)).resolves.toMatchObject({ + serializedOutput: '"hydrated"', + }); + expect(calls).toHaveLength(3); + expect(calls.map((call) => call.metadata.get("attempt"))).toEqual([["1"], ["2"], ["3"]]); + expect(store.download).toHaveBeenCalledTimes(1); + }); + + it("does not retry a metadata deadline or start the wait RPC", async () => { + const failure = Object.assign(new Error("metadata deadline"), { code: grpc.status.DEADLINE_EXCEEDED }); + payloadClient["_metadataGenerator"] = async () => { + throw failure; + }; + await expect(payloadClient.waitForOrchestrationCompletion("grpc-instance", true, 1)).rejects.toBe(failure); + expect(calls).toHaveLength(0); + expect(store.download).not.toHaveBeenCalled(); + }); + + it.each(["abort", "timeout", "stop"] as const)( + "preserves %s cancellation while the successful wait response is hydrating", + async (cause) => { + let started!: () => void; + const downloading = new Promise((resolve) => (started = resolve)); + store.download.mockImplementation( + (_token, signal) => + new Promise((_resolve, reject) => { + signal!.addEventListener("abort", () => reject(signal!.reason), { once: true }); + started(); + }), + ); + const controller = new AbortController(); + const reason = new Error("cancel hydration"); + const pending = payloadClient.waitForOrchestrationCompletion("grpc-instance", true, 0.5, controller.signal); + const rejected = pending.catch((error: unknown) => error); + await downloading; + if (cause === "abort") controller.abort(reason); + if (cause === "stop") { + stopped = true; + await payloadClient.stop(); + } + const error = await rejected; + if (cause === "timeout") expect(error).toBeInstanceOf(TimeoutError); + else expect(error).toBe(cause === "abort" ? reason : payloadClient["_payloadCancellation"].signal.reason); + expect(calls).toHaveLength(1); + expect(store.download).toHaveBeenCalledTimes(1); + expect(store.download.mock.calls[0][1]?.aborted).toBe(true); + }, + ); + }); }); diff --git a/packages/durabletask-js/test/client-wait.spec.ts b/packages/durabletask-js/test/client-wait.spec.ts index e3db56c9..f71f933a 100644 --- a/packages/durabletask-js/test/client-wait.spec.ts +++ b/packages/durabletask-js/test/client-wait.spec.ts @@ -8,6 +8,7 @@ import { TaskHubGrpcClient } from "../src/client/client"; import { OrchestrationState } from "../src/orchestration/orchestration-state"; import { TimeoutError } from "../src/exception/timeout-error"; import { NoOpLogger } from "../src/types/logger.type"; +import { PayloadStore } from "../src/payloads/payload-store"; import * as pb from "../src/proto/orchestrator_service_pb"; type Wait = ( @@ -33,6 +34,152 @@ function completedResponse(): pb.GetInstanceResponse { return response; } +class UnaryWaitPayloadStore extends PayloadStore { + upload = jest.fn(async (_value: string, _signal?: AbortSignal) => "payload:input"); + download = jest.fn(async (_token: string, _signal?: AbortSignal) => '"done"'); + isKnownPayloadToken(value: string): boolean { + return value.startsWith("payload:"); + } +} + +describe("payload-enabled unary completion pipeline", () => { + let client: TaskHubGrpcClient; + let store: UnaryWaitPayloadStore; + let callback: Callback; + const cancel = jest.fn(); + const call: grpc.ClientUnaryCall = Object.assign(new EventEmitter(), { + cancel, + getPeer: () => "test", + getAuthContext: () => null, + }); + + const rpc = jest.fn(); + const metadataGenerator = jest.fn, []>(); + + beforeEach(() => { + jest.useFakeTimers(); + cancel.mockReset(); + store = new UnaryWaitPayloadStore(); + metadataGenerator.mockReset().mockImplementation(async () => { + const metadata = new grpc.Metadata(); + metadata.set("attempt", String(metadataGenerator.mock.calls.length)); + return metadata; + }); + rpc.mockReset().mockImplementation((_req, _metadata, cb) => { + callback = cb; + return call; + }); + client = new TaskHubGrpcClient({ + payloads: { store, thresholdBytes: 1 }, + metadataGenerator, + logger: new NoOpLogger(), + }); + Object.defineProperty(client["_stub"], "waitForInstanceCompletion", { value: rpc }); + }); + + afterEach(() => { + client["_stub"].close(); + jest.useRealTimers(); + }); + + it("does not retry a frozen metadata deadline after a real RPC retry", async () => { + const failure = Object.freeze(grpcError(grpc.status.DEADLINE_EXCEEDED)); + metadataGenerator.mockResolvedValueOnce(new grpc.Metadata()).mockRejectedValue(failure); + const pending = client.waitForOrchestrationCompletion("instance", true, 5); + const assertion = expect(pending).rejects.toBe(failure); + await jest.advanceTimersByTimeAsync(0); + callback(grpcError(grpc.status.DEADLINE_EXCEEDED), new pb.GetInstanceResponse()); + await jest.advanceTimersByTimeAsync(100); + await assertion; + await jest.advanceTimersByTimeAsync(1000); + + expect(metadataGenerator).toHaveBeenCalledTimes(2); + expect(rpc).toHaveBeenCalledTimes(1); + expect(store.upload).not.toHaveBeenCalled(); + expect(store.download).not.toHaveBeenCalled(); + expect(jest.getTimerCount()).toBe(0); + }); + + it("reuses the request across RPC retries, regenerates metadata, and hydrates only the successful response", async () => { + const pending = client.waitForOrchestrationCompletion("instance", true, 5); + await jest.advanceTimersByTimeAsync(0); + const response = completedResponse(); + response.getOrchestrationstate()!.setOutput(new StringValue().setValue("payload:output")); + callback(grpcError(grpc.status.DEADLINE_EXCEEDED), response); + await jest.advanceTimersByTimeAsync(100); + callback(grpcError(grpc.status.DEADLINE_EXCEEDED), response); + await jest.advanceTimersByTimeAsync(200); + callback(null, response); + await expect(pending).resolves.toMatchObject({ serializedOutput: '"done"' }); + + expect(rpc).toHaveBeenCalledTimes(3); + expect(rpc.mock.calls.every(([request]) => request === rpc.mock.calls[0][0])).toBe(true); + expect(rpc.mock.calls.map(([, metadata]) => metadata.get("attempt"))).toEqual([["1"], ["2"], ["3"]]); + expect(store.upload).not.toHaveBeenCalled(); + expect(store.download).toHaveBeenCalledTimes(1); + expect(jest.getTimerCount()).toBe(0); + }); + + it.each(["metadata", "rpc", "backoff", "hydration"] as const)( + "stop cancels %s without starting later metadata or RPC attempts", + async (stage) => { + let resolveMetadata: ((metadata: grpc.Metadata) => void) | undefined; + if (stage === "metadata") { + metadataGenerator.mockImplementation(() => new Promise((resolve) => (resolveMetadata = resolve))); + } + if (stage === "hydration") { + store.download.mockImplementation( + (_token, signal) => + new Promise((_resolve, reject) => { + signal!.addEventListener("abort", () => reject(signal!.reason), { once: true }); + }), + ); + } + const pending = client.waitForOrchestrationCompletion("instance", true, 5); + const rejected = pending.catch((error: unknown) => error); + await jest.advanceTimersByTimeAsync(0); + if (stage === "backoff") { + callback(grpcError(grpc.status.DEADLINE_EXCEEDED), new pb.GetInstanceResponse()); + } else if (stage === "hydration") { + const response = completedResponse(); + response.getOrchestrationstate()!.setOutput(new StringValue().setValue("payload:output")); + callback(null, response); + } + await jest.advanceTimersByTimeAsync(0); + const stopping = client.stop(); + resolveMetadata?.(new grpc.Metadata()); + await jest.advanceTimersByTimeAsync(10000); + await stopping; + + expect(await rejected).toBe(client["_payloadCancellation"].signal.reason); + expect(metadataGenerator).toHaveBeenCalledTimes(1); + expect(rpc).toHaveBeenCalledTimes(stage === "metadata" ? 0 : 1); + expect(cancel).toHaveBeenCalledTimes(stage === "rpc" ? 1 : 0); + expect(store.download).toHaveBeenCalledTimes(stage === "hydration" ? 1 : 0); + expect(jest.getTimerCount()).toBe(0); + }, + ); + + it.each(["abort", "timeout"] as const)("does not restart after %s in payload-enabled backoff", async (cause) => { + const controller = new AbortController(); + const reason = Object.freeze(grpcError(grpc.status.DEADLINE_EXCEEDED)); + const pending = client.waitForOrchestrationCompletion("instance", true, 0.05, controller.signal); + const rejected = pending.catch((error: unknown) => error); + await jest.advanceTimersByTimeAsync(0); + callback(grpcError(grpc.status.DEADLINE_EXCEEDED), new pb.GetInstanceResponse()); + await jest.advanceTimersByTimeAsync(0); + if (cause === "abort") controller.abort(reason); + await jest.advanceTimersByTimeAsync(10000); + + if (cause === "abort") expect(await rejected).toBe(reason); + else expect(await rejected).toBeInstanceOf(TimeoutError); + expect(metadataGenerator).toHaveBeenCalledTimes(1); + expect(rpc).toHaveBeenCalledTimes(1); + expect(cancel).not.toHaveBeenCalled(); + expect(jest.getTimerCount()).toBe(0); + }); +}); + describe.each([ ["waitForOrchestrationStart", "waitForInstanceStart", false, "start"], ["waitForOrchestrationCompletion", "waitForInstanceCompletion", true, "complete"], diff --git a/packages/durabletask-js/test/functions-grpc-support.spec.ts b/packages/durabletask-js/test/functions-grpc-support.spec.ts index c6a55717..293c98a5 100644 --- a/packages/durabletask-js/test/functions-grpc-support.spec.ts +++ b/packages/durabletask-js/test/functions-grpc-support.spec.ts @@ -3,6 +3,8 @@ import { OrchestrationContext, + PayloadStore, + PayloadStorageException, TaskEntity, TaskHubGrpcWorker, TOrchestrator, @@ -12,6 +14,7 @@ import { 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"; @@ -88,6 +91,50 @@ describe("Functions gRPC support surface", () => { expect(response.getResultsList()[0].getSuccess()?.getResult()?.getValue()).toBe("1"); expect(response.getEntitystate()?.getValue()).toBe("1"); }); + + it.each([ + { kind: "orchestrator", phase: "download" }, + { kind: "entity", phase: "download" }, + { kind: "orchestrator", phase: "upload" }, + { kind: "entity", phase: "upload" }, + { kind: "entity", phase: "maximum" }, + ])("preserves the original $phase failure for a tokenless $kind host invocation", async ({ kind, phase }) => { + const failure = Object.assign(new Error("storage unavailable"), { statusCode: 503 }); + class Store extends PayloadStore { + async upload(): Promise { + throw failure; + } + async download(): Promise { + throw failure; + } + isKnownPayloadToken(value: string): boolean { + return value.startsWith("payload:"); + } + } + const worker = new TaskHubGrpcWorker({ + payloads: { store: new Store(), thresholdBytes: 1, maxPayloadBytes: phase === "maximum" ? 1 : undefined }, + logger: new NoOpLogger(), + }); + worker.addNamedOrchestrator("flow", async () => "output"); + worker.addNamedEntity("counter", () => new CounterEntity()); + const input = new StringValue().setValue(phase === "download" ? "payload:input" : "10"); + const request = + kind === "orchestrator" + ? new pb.OrchestratorRequest() + .setInstanceid(TEST_INSTANCE_ID) + .setNeweventsList([ + newOrchestratorStartedEvent(), + newExecutionStartedEvent("flow", TEST_INSTANCE_ID, input.getValue()), + ]) + : createEntityBatchRequest("counter", "key").setEntitystate(input); + const processing = + kind === "orchestrator" + ? worker.processOrchestratorRequest(request.serializeBinary()) + : worker.processEntityBatchRequest(request.serializeBinary()); + if (phase === "maximum") await expect(processing).rejects.toBeInstanceOf(PayloadStorageException); + else await expect(processing).rejects.toBe(failure); + expect(worker["_pendingWorkItems"].size).toBe(0); + }); }); function createEntityBatchRequest(entityName: string, entityKey: string): pb.EntityBatchRequest { 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 00000000..6dea7e05 --- /dev/null +++ b/packages/durabletask-js/test/payload-transport.spec.ts @@ -0,0 +1,560 @@ +// 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, + TaskEntity, + OrchestrationAlreadyExistsError, + OrchestrationStatus, +} 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); +type WorkKind = "activity" | "orchestrator" | "streamed-orchestrator" | "entity-v1" | "entity-v2"; + +function workItem(kind: WorkKind, input: string, token: string): pb.WorkItem { + const item = new pb.WorkItem().setCompletiontoken(token); + if (kind === "activity") { + return item.setActivityrequest( + new pb.ActivityRequest() + .setName("echo") + .setTaskid(1) + .setInput(text(input)) + .setOrchestrationinstance(new pb.OrchestrationInstance().setInstanceid("instance")), + ); + } + if (kind === "orchestrator" || kind === "streamed-orchestrator") { + return item.setOrchestratorrequest( + new pb.OrchestratorRequest() + .setInstanceid("instance") + .setRequireshistorystreaming(kind === "streamed-orchestrator") + .setNeweventsList( + kind === "streamed-orchestrator" + ? [pbh.newOrchestratorStartedEvent()] + : [pbh.newOrchestratorStartedEvent(), pbh.newExecutionStartedEvent("flow", "instance", input)], + ), + ); + } + if (kind === "entity-v1") { + return item.setEntityrequest( + new pb.EntityBatchRequest() + .setInstanceid("@echo@key") + .setOperationsList([ + new pb.OperationRequest().setOperation("echo").setRequestid("request").setInput(text(input)), + ]), + ); + } + return item.setEntityrequestv2( + new pb.EntityRequest() + .setInstanceid("@echo@key") + .setOperationrequestsList([ + new pb.HistoryEvent().setEntityoperationcalled( + new pb.EntityOperationCalledEvent().setOperation("echo").setRequestid("request").setInput(text(input)), + ), + ]), + ); +} + +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 activityResponses: pb.ActivityResponse[]; + let entityResponses: pb.EntityBatchResult[]; + let abandonments: { kind: string; token: string }[]; + let onCompleteActivity: stubs.ITaskHubSidecarServiceServer["completeActivityTask"]; + let errors: jest.Mock; + let historyEvents: pb.HistoryEvent[]; + let startCalls: grpc.ServerUnaryCall[]; + let onStartInstance: stubs.ITaskHubSidecarServiceServer["startInstance"]; + let clientStopped: boolean; + + beforeEach(async () => { + store = new MemoryStore(); + input = await store.upload(JSON.stringify({ message: "hydrated" })); + subscription = undefined; + activityResponse = undefined; + activityCalls = 0; + requestedInput = undefined; + orchestrationResponses = []; + activityResponses = []; + entityResponses = []; + abandonments = []; + errors = jest.fn(); + clientStopped = false; + startCalls = []; + onStartInstance = (call, callback) => + callback(null, new pb.CreateInstanceResponse().setInstanceid(call.request.getInstanceid())); + onCompleteActivity = (_call, callback) => callback(null, new pb.CompleteTaskResponse()); + 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) => { + startCalls.push(call); + requestedInput = call.request.getInput()?.getValue(); + onStartInstance(call, callback); + }, + 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; + activityResponses.push(call.request); + onCompleteActivity(call, callback); + }, + completeOrchestratorTask: (call, callback) => { + orchestrationResponses.push(call.request); + callback(null, new pb.CompleteTaskResponse()); + }, + completeEntityTask: (call, callback) => { + entityResponses.push(call.request); + callback(null, new pb.CompleteTaskResponse()); + }, + abandonTaskActivityWorkItem: (call, callback) => { + abandonments.push({ kind: "activity", token: call.request.getCompletiontoken() }); + callback(null, new pb.AbandonActivityTaskResponse()); + }, + abandonTaskOrchestratorWorkItem: (call, callback) => { + abandonments.push({ kind: "orchestrator", token: call.request.getCompletiontoken() }); + callback(null, new pb.AbandonOrchestrationTaskResponse()); + }, + abandonTaskEntityWorkItem: (call, callback) => { + abandonments.push({ kind: "entity", token: call.request.getCompletiontoken() }); + callback(null, new pb.AbandonEntityTaskResponse()); + }, + } satisfies Pick< + stubs.ITaskHubSidecarServiceServer, + | "hello" + | "startInstance" + | "getInstance" + | "waitForInstanceStart" + | "queryInstances" + | "streamInstanceHistory" + | "getWorkItems" + | "completeActivityTask" + | "completeOrchestratorTask" + | "completeEntityTask" + | "abandonTaskActivityWorkItem" + | "abandonTaskOrchestratorWorkItem" + | "abandonTaskEntityWorkItem" + >; + 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(); + if (!clientStopped) await client.stop(); + server.forceShutdown(); + jest.restoreAllMocks(); + }); + + function registerPayloadWorker(output?: unknown, maxPayloadBytes?: number): jest.Mock { + worker = new TaskHubGrpcWorker({ + hostAddress, + payloads: { store, thresholdBytes: 1, maxPayloadBytes }, + logger: { error: errors, warn: () => {}, info: () => {}, debug: () => {} }, + }); + const invoked = jest.fn((value: unknown) => output ?? value); + worker.addNamedActivity("echo", async (_context, value) => invoked(value)); + worker.addNamedOrchestrator("flow", async (_context, value) => invoked(value)); + class Echo extends TaskEntity { + echo(value: unknown) { + return invoked(value); + } + protected initializeState(): null { + return null; + } + } + worker.addNamedEntity("echo", () => new Echo()); + return invoked; + } + + const completeCount = () => activityResponses.length + orchestrationResponses.length + entityResponses.length; + + 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"}' }); + }); + + describe("scheduling error domains", () => { + it.each([grpc.status.ALREADY_EXISTS, grpc.status.INVALID_ARGUMENT])( + "preserves upload status %s without scheduling", + async (code) => { + const failure = Object.assign(new Error("storage rejected upload"), { + code, + details: "storage rejected upload", + metadata: new grpc.Metadata(), + }); + const upload = jest.spyOn(store, "upload").mockRejectedValue(failure); + const metadata = jest.fn(async () => new grpc.Metadata()); + client["_metadataGenerator"] = metadata; + + await expect(client.scheduleNewOrchestration("workflow", { message: "request" })).rejects.toBe(failure); + expect(upload).toHaveBeenCalledTimes(1); + expect(metadata).not.toHaveBeenCalled(); + expect(startCalls).toHaveLength(0); + }, + ); + + describe.each([true, false])("with payloads enabled=%s", (payloadsEnabled) => { + beforeEach(async () => { + await client.stop(); + client = new TaskHubGrpcClient({ + hostAddress, + payloads: payloadsEnabled ? { store, thresholdBytes: 1 } : undefined, + defaultVersion: "default-version", + logger: new NoOpLogger(), + }); + }); + + it.each([grpc.status.ALREADY_EXISTS, grpc.status.INVALID_ARGUMENT])( + "preserves metadata status %s without scheduling", + async (code) => { + const failure = Object.assign(new Error("metadata rejected request"), { code }); + const upload = jest.spyOn(store, "upload"); + const metadata = jest.fn(async () => { + throw failure; + }); + client["_metadataGenerator"] = metadata; + + await expect(client.scheduleNewOrchestration("workflow", { message: "request" })).rejects.toBe(failure); + expect(metadata).toHaveBeenCalledTimes(1); + expect(upload).toHaveBeenCalledTimes(payloadsEnabled ? 1 : 0); + expect(startCalls).toHaveLength(0); + }, + ); + + it.each([ + [grpc.status.ALREADY_EXISTS, OrchestrationAlreadyExistsError], + [grpc.status.INVALID_ARGUMENT, TypeError], + ])("translates actual startInstance status %s with its cause and details", async (code, expectedError) => { + onStartInstance = (_call, callback) => callback({ code, details: "scheduler rejected start" }); + + const scheduled = client.scheduleNewOrchestration("workflow", { message: "request" }); + await expect(scheduled).rejects.toBeInstanceOf(expectedError); + await expect(scheduled).rejects.toMatchObject({ + message: "scheduler rejected start", + cause: expect.objectContaining({ code, details: "scheduler rejected start" }), + }); + expect(startCalls).toHaveLength(1); + }); + + it.each([undefined, "explicit-version"])( + "preserves successful scheduling options (version=%s)", + async (version) => { + const metadata = new grpc.Metadata(); + metadata.set("taskhub", "scheduling-error-domains"); + const generateMetadata = jest.fn(async () => metadata); + client["_metadataGenerator"] = generateMetadata; + onStartInstance = (_call, callback) => + callback(null, new pb.CreateInstanceResponse().setInstanceid("scheduled-instance")); + const upload = jest.spyOn(store, "upload"); + const startAt = new Date("2026-10-22T15:00:00.000Z"); + const input = { message: "request \u4f60\u597d" }; + + await expect( + client.scheduleNewOrchestration("workflow", input, { + instanceId: "requested-instance", + startAt, + tags: { owner: "scheduling-error-domains" }, + version, + dedupeStatuses: [OrchestrationStatus.COMPLETED], + }), + ).resolves.toBe("scheduled-instance"); + + expect(startCalls).toHaveLength(1); + const { request, metadata: receivedMetadata } = startCalls[0]; + expect(request.getName()).toBe("workflow"); + expect(request.getInstanceid()).toBe("requested-instance"); + expect(request.getScheduledstarttimestamp()?.toDate()).toEqual(startAt); + expect(request.getVersion()?.getValue()).toBe(version ?? "default-version"); + expect(request.getTagsMap().get("owner")).toBe("scheduling-error-domains"); + expect(request.getOrchestrationidreusepolicy()?.getReplaceablestatusList()).toHaveLength(6); + expect(request.getOrchestrationidreusepolicy()?.getReplaceablestatusList()).not.toContain( + pb.OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED, + ); + expect(receivedMetadata.get("taskhub")).toEqual(["scheduling-error-domains"]); + expect(generateMetadata).toHaveBeenCalledTimes(1); + expect(upload).toHaveBeenCalledTimes(payloadsEnabled ? 1 : 0); + const wireInput = request.getInput()!.getValue(); + expect(payloadsEnabled ? await store.download(wireInput) : wireInput).toBe(JSON.stringify(input)); + }, + ); + }); + + it("preserves stop cancellation during upload without starting the scheduling RPC", async () => { + let uploadSignal: AbortSignal | undefined; + const upload = jest.spyOn(store, "upload").mockImplementation( + (_value: string, signal?: AbortSignal) => + new Promise((_resolve, reject) => { + uploadSignal = signal; + signal!.addEventListener("abort", () => reject(signal!.reason), { once: true }); + }), + ); + const pending = client.scheduleNewOrchestration("workflow", { message: "request" }); + const rejected = pending.catch((error: unknown) => error); + await waitFor(() => !!uploadSignal); + clientStopped = true; + await client.stop(); + + expect(await rejected).toBe(uploadSignal!.reason); + expect(uploadSignal!.aborted).toBe(true); + expect(upload).toHaveBeenCalledTimes(1); + expect(startCalls).toHaveLength(0); + }); + + it("preserves stop cancellation of the pending scheduling RPC", async () => { + let canceled = false; + onStartInstance = (call) => { + call.on("cancelled", () => { + canceled = true; + }); + }; + const pending = client.scheduleNewOrchestration("workflow", { message: "request" }); + const rejected = pending.catch((error: unknown) => error); + await waitFor(() => startCalls.length === 1); + clientStopped = true; + await client.stop(); + await waitFor(() => canceled); + + expect(await rejected).toBe(client["_payloadCancellation"].signal.reason); + expect(startCalls).toHaveLength(1); + }); + }); + + 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(abandonments).toEqual([]); + 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.each(["activity", "orchestrator", "streamed-orchestrator", "entity-v1", "entity-v2"])( + "abandons %s on download failure and permits redelivery after storage recovers", + async (kind) => { + const invoked = registerPayloadWorker(); + const failure = Object.assign(new Error("download unavailable"), { statusCode: 503 }); + const download = jest.spyOn(store, "download").mockRejectedValueOnce(failure); + await worker.start(); + await waitFor(() => !!subscription); + const stream = worker["_responseStream"]; + subscription!.write(workItem(kind, input, "retryable")); + await waitFor(() => errors.mock.calls.length > 0 && worker["_pendingWorkItems"].size === 0); + expect(invoked).not.toHaveBeenCalled(); + expect(download).toHaveBeenCalledTimes(1); + expect(completeCount()).toBe(0); + expect(abandonments).toEqual([ + { + kind: kind.startsWith("entity") ? "entity" : kind === "activity" ? "activity" : "orchestrator", + token: "retryable", + }, + ]); + expect(subscription!.cancelled).toBe(false); + expect(worker["_responseStream"]).toBe(stream); + + subscription!.write(workItem(kind, input, "redelivered")); + await waitFor(() => completeCount() === 1 && worker["_pendingWorkItems"].size === 0); + expect(download).toHaveBeenCalledTimes(2); + expect(invoked).toHaveBeenCalledTimes(1); + expect(invoked).toHaveBeenCalledWith({ message: "hydrated" }); + expect(abandonments).toHaveLength(1); + }, + ); + + it.each<{ kind: WorkKind; maximum?: number }>([ + { kind: "activity" }, + { kind: "orchestrator" }, + { kind: "entity-v1" }, + { kind: "entity-v2" }, + { kind: "entity-v1", maximum: 64 }, + { kind: "entity-v2", maximum: 64 }, + ])("abandons $kind when output externalization fails (maximum=$maximum)", async ({ kind, maximum }) => { + const invoked = registerPayloadWorker("x".repeat(128), maximum); + const upload = jest.spyOn(store, "upload"); + if (!maximum) upload.mockRejectedValueOnce(Object.assign(new Error("upload unavailable"), { statusCode: 503 })); + await worker.start(); + await waitFor(() => !!subscription); + const stream = worker["_responseStream"]; + subscription!.write(workItem(kind, '"inline"', "output-failure")); + await waitFor(() => errors.mock.calls.length > 0 && worker["_pendingWorkItems"].size === 0); + expect(invoked).toHaveBeenCalledTimes(1); + expect(upload).toHaveBeenCalledTimes(maximum ? 0 : 1); + expect(completeCount()).toBe(0); + expect(abandonments).toEqual([ + { + kind: kind.startsWith("entity") ? "entity" : kind === "activity" ? "activity" : "orchestrator", + token: "output-failure", + }, + ]); + expect(subscription!.cancelled).toBe(false); + expect(worker["_responseStream"]).toBe(stream); + }); + + it.each(["activity", "orchestrator"])( + "preserves permanent %s output failure conversion without abandoning", + async (kind) => { + registerPayloadWorker("x".repeat(128), 64); + await worker.start(); + await waitFor(() => !!subscription); + subscription!.write(workItem(kind, '"inline"', "permanent")); + await waitFor(() => completeCount() === 1 && worker["_pendingWorkItems"].size === 0); + const failure = + kind === "activity" + ? activityResponses[0].getFailuredetails() + : orchestrationResponses[0].getActionsList()[0].getCompleteorchestration()?.getFailuredetails(); + expect(failure?.getErrortype()).toBe("PayloadStorageException"); + expect(failure?.getIsnonretriable()).toBe(true); + expect(abandonments).toEqual([]); + }, + ); + + it("retries a transient completion RPC with the same externalized response and only one upload", async () => { + const invoked = registerPayloadWorker("output"); + const upload = jest.spyOn(store, "upload"); + onCompleteActivity = (_call, callback) => { + if (activityResponses.length === 1) callback({ code: grpc.status.UNAVAILABLE, details: "retry completion" }); + else callback(null, new pb.CompleteTaskResponse()); + }; + await worker.start(); + await waitFor(() => !!subscription); + subscription!.write(workItem("activity", '"inline"', "response-retry")); + await waitFor(() => activityResponses.length === 2 && worker["_pendingWorkItems"].size === 0); + expect(invoked).toHaveBeenCalledTimes(1); + expect(upload).toHaveBeenCalledTimes(1); + expect(activityResponses[0].getResult()?.getValue()).toMatch(/^payload:/); + expect(activityResponses[1].serializeBinary()).toEqual(activityResponses[0].serializeBinary()); + expect(abandonments).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); +}); diff --git a/packages/durabletask-js/test/worker-response-delivery.spec.ts b/packages/durabletask-js/test/worker-response-delivery.spec.ts index 7de57a9d..bda5759b 100644 --- a/packages/durabletask-js/test/worker-response-delivery.spec.ts +++ b/packages/durabletask-js/test/worker-response-delivery.spec.ts @@ -8,6 +8,7 @@ import { StringValue } from "google-protobuf/google/protobuf/wrappers_pb"; import { Value } from "google-protobuf/google/protobuf/struct_pb"; import { newExecutionStartedEvent, newOrchestratorStartedEvent } from "../src/utils/pb-helper.util"; import { TaskEntity } from "../src/entities/task-entity"; +import { PayloadStore } from "../src/payloads/payload-store"; import * as pb from "../src/proto/orchestrator_service_pb"; import * as stubs from "../src/proto/orchestrator_service_grpc_pb"; import { TaskHubGrpcWorker } from "../src/worker/task-hub-grpc-worker"; @@ -33,6 +34,14 @@ function activityRequest() { .setOrchestrationinstance(new pb.OrchestrationInstance().setInstanceid("instance")); } +class WorkerPayloadStore extends PayloadStore { + upload = jest.fn(async (_value: string, _signal?: AbortSignal) => "payload:output"); + download = jest.fn(async (_token: string, _signal?: AbortSignal) => '"input"'); + isKnownPayloadToken(value: string): boolean { + return value.startsWith("payload:"); + } +} + describe("Worker response retries", () => { let stub: stubs.TaskHubSidecarServiceClient; let worker: TaskHubGrpcWorker; @@ -64,19 +73,24 @@ describe("Worker response retries", () => { observed(ctx.sourceInstanceId, ctx.instanceId, ctx.version); return "done"; }); - const send = jest.spyOn(stub, "completeOrchestratorTask").mockImplementation( - (_request, _metadata, optionsOrCallback: Partial | Callback, callback?: Callback) => { - const respond = typeof optionsOrCallback === "function" ? optionsOrCallback : callback!; - respond(null, new pb.CompleteTaskResponse()); - return unaryCall(); - }, - ); - const request = new pb.OrchestratorRequest().setInstanceid("clone-id").setNeweventsList([ - newOrchestratorStartedEvent(), - newExecutionStartedEvent("Flow", "clone-id", undefined, undefined, "execution-id", "1.0"), - ]); + const send = jest + .spyOn(stub, "completeOrchestratorTask") + .mockImplementation( + (_request, _metadata, optionsOrCallback: Partial | Callback, callback?: Callback) => { + const respond = typeof optionsOrCallback === "function" ? optionsOrCallback : callback!; + respond(null, new pb.CompleteTaskResponse()); + return unaryCall(); + }, + ); + const request = new pb.OrchestratorRequest() + .setInstanceid("clone-id") + .setNeweventsList([ + newOrchestratorStartedEvent(), + newExecutionStartedEvent("Flow", "clone-id", undefined, undefined, "execution-id", "1.0"), + ]); request.setExecutionid(new StringValue().setValue("execution-id")); - if (sourceInstanceId !== undefined) request.getPropertiesMap().set("sourceInstanceId", Value.fromJavaScript(sourceInstanceId)); + if (sourceInstanceId !== undefined) + request.getPropertiesMap().set("sourceInstanceId", Value.fromJavaScript(sourceInstanceId)); const item = new pb.WorkItem().setCompletiontoken("token").setOrchestratorrequest(request); worker["_dispatchWorkItem"](pb.WorkItem.deserializeBinary(item.serializeBinary()), stub); await Promise.all(worker["_pendingWorkItems"]); @@ -391,4 +405,221 @@ describe("Worker response retries", () => { await stoppingNewRun; } }); + + describe("payload failure release", () => { + let store: WorkerPayloadStore; + let abandon: jest.SpiedFunction; + let complete: jest.SpiedFunction; + let onAbandon: (request: pb.AbandonActivityTaskRequest, callback: Callback) => void; + let cancel: jest.Mock; + const failure = Object.assign(new Error("payload storage unavailable"), { statusCode: 503 }); + + beforeEach(() => { + store = new WorkerPayloadStore(); + worker = new TaskHubGrpcWorker({ payloads: { store, thresholdBytes: 1 }, logger, shutdownTimeoutMs: 100 }); + worker["_abortController"] = controller; + worker["_stub"] = stub; + worker.addNamedActivity("answer", () => 42); + cancel = jest.fn(); + onAbandon = (_request, callback) => callback(null, new pb.AbandonActivityTaskResponse()); + abandon = jest + .spyOn(stub, "abandonTaskActivityWorkItem") + .mockImplementation( + ( + request, + _metadata, + optionsOrCallback: Partial | Callback, + callback?: Callback, + ) => { + onAbandon(request, typeof optionsOrCallback === "function" ? optionsOrCallback : callback!); + return unaryCall(cancel); + }, + ); + complete = jest + .spyOn(stub, "completeActivityTask") + .mockImplementation( + (_request, _metadata, optionsOrCallback: Partial | Callback, callback?: Callback) => { + (typeof optionsOrCallback === "function" ? optionsOrCallback : callback!)( + null, + new pb.CompleteTaskResponse(), + ); + return unaryCall(); + }, + ); + }); + + function dispatch(): void { + worker["_dispatchWorkItem"]( + new pb.WorkItem() + .setCompletiontoken("payload-token") + .setActivityrequest(activityRequest().setInput(new StringValue().setValue("payload:input"))), + stub, + ); + } + + it("awaits abandon and retries transient release errors with the same token and fresh metadata", async () => { + store.download.mockRejectedValueOnce(failure); + const metadata = jest.fn(async () => new grpc.Metadata()); + worker["_metadataGenerator"] = metadata; + let finish!: () => void; + onAbandon = (_request, callback) => { + if (abandon.mock.calls.length === 1) + callback(grpcError(grpc.status.UNAVAILABLE), new pb.AbandonActivityTaskResponse()); + else finish = () => callback(null, new pb.AbandonActivityTaskResponse()); + }; + dispatch(); + await jest.advanceTimersByTimeAsync(200); + expect(abandon).toHaveBeenCalledTimes(2); + expect(worker["_pendingWorkItems"].size).toBe(1); + expect(abandon.mock.calls[0][0].getCompletiontoken()).toBe("payload-token"); + expect(abandon.mock.calls[1][0]).toBe(abandon.mock.calls[0][0]); + expect(abandon.mock.calls[1][1]).not.toBe(abandon.mock.calls[0][1]); + finish(); + await Promise.all(worker["_pendingWorkItems"]); + expect(metadata).toHaveBeenCalledTimes(2); + expect(store.download).toHaveBeenCalledTimes(1); + expect(complete).not.toHaveBeenCalled(); + expect(worker["_pendingWorkItems"].size).toBe(0); + expect(logger.error).toHaveBeenCalledTimes(1); + }); + + it.each([grpc.status.PERMISSION_DENIED, grpc.status.UNAVAILABLE])( + "logs terminal abandon failure %s without an unhandled rejection", + async (status) => { + store.download.mockRejectedValueOnce(failure); + onAbandon = (_request, callback) => callback(grpcError(status), new pb.AbandonActivityTaskResponse()); + dispatch(); + await jest.runAllTimersAsync(); + await Promise.all(worker["_pendingWorkItems"]); + expect(abandon).toHaveBeenCalledTimes(status === grpc.status.UNAVAILABLE ? 10 : 1); + expect(logger.error).toHaveBeenCalledWith(expect.stringContaining("payload storage unavailable")); + expect(logger.error).toHaveBeenCalledWith(expect.stringContaining("delivery failed")); + expect(complete).not.toHaveBeenCalled(); + expect(worker["_pendingWorkItems"].size).toBe(0); + expect(jest.getTimerCount()).toBe(0); + }, + ); + + it("releases through the captured stub even if the worker replaces its current stub", async () => { + let reject!: (error: Error) => void; + store.download.mockReturnValueOnce(new Promise((_resolve, fail) => (reject = fail))); + const replacement = new stubs.TaskHubSidecarServiceClient("localhost:2", grpc.credentials.createInsecure()); + const replacementAbandon = jest.spyOn(replacement, "abandonTaskActivityWorkItem"); + try { + dispatch(); + await jest.advanceTimersByTimeAsync(0); + worker["_stub"] = replacement; + reject(failure); + await jest.runAllTimersAsync(); + await Promise.all(worker["_pendingWorkItems"]); + expect(abandon).toHaveBeenCalledTimes(1); + expect(replacementAbandon).not.toHaveBeenCalled(); + expect(complete).not.toHaveBeenCalled(); + } finally { + replacement.close(); + } + }); + + it.each(["download", "upload"] as const)( + "does not abandon a late %s failure with a replacement run's signal", + async (phase) => { + let reject!: (error: Error) => void; + store[phase].mockReturnValueOnce(new Promise((_resolve, fail) => (reject = fail))); + dispatch(); + await jest.advanceTimersByTimeAsync(0); + expect(store[phase]).toHaveBeenCalledTimes(1); + controller.abort(); + worker["_abortController"] = new AbortController(); + reject(failure); + await jest.runAllTimersAsync(); + await Promise.all(worker["_pendingWorkItems"]); + expect(abandon).not.toHaveBeenCalled(); + expect(complete).not.toHaveBeenCalled(); + expect(worker["_pendingWorkItems"].size).toBe(0); + expect(worker["_abortController"]!.signal.aborted).toBe(false); + }, + ); + + it.each(["metadata", "retry", "RPC"] as const)( + "shutdown cancels abandon while awaiting %s without sending in a new run", + async (phase) => { + store.download.mockRejectedValueOnce(failure); + let finishMetadata!: (value: grpc.Metadata) => void; + const metadata = jest.fn(async () => new grpc.Metadata()); + if (phase === "metadata") + metadata.mockImplementation(() => new Promise((resolve) => (finishMetadata = resolve))); + worker["_metadataGenerator"] = metadata; + onAbandon = (_request, callback) => { + if (phase === "retry") callback(grpcError(grpc.status.UNAVAILABLE), new pb.AbandonActivityTaskResponse()); + }; + worker["_isRunning"] = true; + dispatch(); + await jest.advanceTimersByTimeAsync(0); + expect(metadata).toHaveBeenCalledTimes(1); + const stopping = worker.stop(); + await jest.advanceTimersByTimeAsync(1100); + await stopping; + worker["_abortController"] = new AbortController(); + if (phase === "metadata") finishMetadata(new grpc.Metadata()); + await jest.runAllTimersAsync(); + await Promise.all(worker["_pendingWorkItems"]); + expect(abandon).toHaveBeenCalledTimes(phase === "metadata" ? 0 : 1); + expect(cancel).toHaveBeenCalledTimes(phase === "RPC" ? 1 : 0); + expect(complete).not.toHaveBeenCalled(); + expect(worker["_pendingWorkItems"].size).toBe(0); + expect(jest.getTimerCount()).toBe(0); + }, + ); + + it.each([undefined, grpc.status.PERMISSION_DENIED])( + "does not double abandon an explicit version rejection (release status=%s)", + async (status) => { + worker = new TaskHubGrpcWorker({ + payloads: { store }, + logger, + versioning: { + version: "2", + matchStrategy: VersionMatchStrategy.Strict, + failureStrategy: VersionFailureStrategy.Reject, + }, + }); + worker["_abortController"] = controller; + const reject = jest + .spyOn(stub, "abandonTaskOrchestratorWorkItem") + .mockImplementation( + ( + _request, + _metadata, + optionsOrCallback: Partial | Callback, + callback?: Callback, + ) => { + (typeof optionsOrCallback === "function" ? optionsOrCallback : callback!)( + status === undefined ? null : grpcError(status), + new pb.AbandonOrchestrationTaskResponse(), + ); + return unaryCall(); + }, + ); + worker["_dispatchWorkItem"]( + new pb.WorkItem() + .setCompletiontoken("version-token") + .setOrchestratorrequest( + new pb.OrchestratorRequest() + .setInstanceid("instance") + .setNeweventsList([ + newOrchestratorStartedEvent(), + newExecutionStartedEvent("flow", "instance", undefined, undefined, undefined, "1"), + ]), + ), + stub, + ); + await jest.runAllTimersAsync(); + await Promise.all(worker["_pendingWorkItems"]); + expect(reject).toHaveBeenCalledTimes(1); + expect(reject.mock.calls[0][0].getCompletiontoken()).toBe("version-token"); + expect(abandon).not.toHaveBeenCalled(); + expect(complete).not.toHaveBeenCalled(); + }, + ); + }); });