diff --git a/CHANGELOG.md b/CHANGELOG.md index 91ac550..1040a15 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,10 @@ ### New +- Add payload storage contracts and bounded protobuf payload processing, with inclusive UTF-8 + thresholds, explicit storage/size failures, and non-retriable worker upload-failure conversion. + This is phase 1 of the external-payload feature; transport, providers, and cleanup follow separately. + Reserve core `0.4.1` for the new APIs; do not publish partial feature phases. - Expose readonly `OrchestrationContext.sourceInstanceId` from the existing protobuf request properties on initial execution and replay, independently of parent identity. Strings are preserved verbatim; null or missing metadata returns `undefined`. @@ -43,6 +47,12 @@ ### Fixes +- 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 + and token mutations for deterministically oversized messages. Precancellation, worker failure + conversion, and bounded I/O error handling are preserved; transient I/O or submission failures + can still leave stored payloads. - Prevent shared or duplicate pending tasks from stranding `whenAll` orchestrations after all activities finish. Notify every dependent composite, preserve duplicate result positions, and detach completed `whenAny` groups from their remaining children. diff --git a/package-lock.json b/package-lock.json index 8ef8ff3..e1d0443 100644 --- a/package-lock.json +++ b/package-lock.json @@ -7600,7 +7600,7 @@ "dependencies": { "@azure/functions": "^4.16.1", "@grpc/grpc-js": "^1.14.4", - "@microsoft/durabletask-js": "0.4.0" + "@microsoft/durabletask-js": "0.4.1" }, "devDependencies": { "@types/jest": "^29.5.1", @@ -7623,7 +7623,7 @@ }, "packages/durabletask-js": { "name": "@microsoft/durabletask-js", - "version": "0.4.0", + "version": "0.4.1", "license": "MIT", "dependencies": { "@grpc/grpc-js": "^1.14.4", diff --git a/packages/azure-functions-durable/package.json b/packages/azure-functions-durable/package.json index d4cf005..20e54be 100644 --- a/packages/azure-functions-durable/package.json +++ b/packages/azure-functions-durable/package.json @@ -68,7 +68,7 @@ "dependencies": { "@azure/functions": "^4.16.1", "@grpc/grpc-js": "^1.14.4", - "@microsoft/durabletask-js": "0.4.0" + "@microsoft/durabletask-js": "0.4.1" }, "peerDependencies": { "@azure/identity": "^4.0.0" diff --git a/packages/durabletask-js/package.json b/packages/durabletask-js/package.json index 598a96c..1b33d65 100644 --- a/packages/durabletask-js/package.json +++ b/packages/durabletask-js/package.json @@ -1,6 +1,6 @@ { "name": "@microsoft/durabletask-js", - "version": "0.4.0", + "version": "0.4.1", "description": "A Durable Task Javascript SDK compatible with Dapr Workflow and its underlying Durable Task engine", "main": "./dist/index.js", "types": "./dist/index.d.ts", diff --git a/packages/durabletask-js/src/index.ts b/packages/durabletask-js/src/index.ts index 58e0275..5e22ec8 100644 --- a/packages/durabletask-js/src/index.ts +++ b/packages/durabletask-js/src/index.ts @@ -2,6 +2,12 @@ // Licensed under the MIT License. // Client and Worker +export { + PayloadStore, + PayloadStorageOptions, + PayloadStorageException, + PayloadDeleteOutcome, +} from "./payloads/payload-store"; export { TaskHubGrpcClient, TaskHubGrpcClientOptions, MetadataGenerator } from "./client/client"; export { TaskHubGrpcWorker, TaskHubGrpcWorkerOptions } from "./worker/task-hub-grpc-worker"; export { ConcurrencyOptions } from "./worker/concurrency-options"; diff --git a/packages/durabletask-js/src/payloads/payload-processor.ts b/packages/durabletask-js/src/payloads/payload-processor.ts new file mode 100644 index 0000000..143ad5b --- /dev/null +++ b/packages/durabletask-js/src/payloads/payload-processor.ts @@ -0,0 +1,205 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { StringValue } from "google-protobuf/google/protobuf/wrappers_pb"; +import * as pb from "../proto/orchestrator_service_pb"; +import { newFailureDetails } from "../utils/pb-helper.util"; +import { PayloadStorageException, PayloadStorageOptions } from "./payload-store"; +import { runBounded } from "./run-bounded"; + +/** Explicit protobuf field mapping based on the .NET AzureBlobPayloads SideCar interceptor. */ +export class PayloadProcessor { + private readonly _threshold: number; + private readonly _maximum: number; + + constructor(private readonly _options: PayloadStorageOptions) { + this._threshold = _options.thresholdBytes ?? 256 * 1024; + this._maximum = _options.maxPayloadBytes ?? 10 * 1024 * 1024; + if (!Number.isSafeInteger(this._threshold) || this._threshold < 0 || this._threshold > 1024 * 1024) { + throw new RangeError("thresholdBytes must be an integer between 0 and 1 MiB."); + } + if (!Number.isSafeInteger(this._maximum) || this._maximum <= 0) { + throw new RangeError("maxPayloadBytes must be a positive safe integer."); + } + if (!_options.store) throw new TypeError("A payload store is required."); + } + + async externalize(message: unknown, signal?: AbortSignal): Promise { + const fields: (StringValue | undefined)[] = []; + if ( + message instanceof pb.CreateInstanceRequest || + message instanceof pb.RaiseEventRequest || + message instanceof pb.SignalEntityRequest + ) { + fields.push(message.getInput()); + } else if (message instanceof pb.TerminateRequest) { + fields.push(message.getOutput()); + } else if ( + message instanceof pb.SuspendRequest || + message instanceof pb.ResumeRequest || + message instanceof pb.RewindInstanceRequest + ) { + fields.push(message.getReason()); + } else if (message instanceof pb.ActivityResponse) { + fields.push(message.getResult()); + } else if (message instanceof pb.OrchestratorResponse) { + fields.push(message.getCustomstatus()); + for (const action of message.getActionsList()) { + fields.push( + action.getScheduletask()?.getInput(), + action.getCreatesuborchestration()?.getInput(), + action.getCompleteorchestration()?.getResult(), + action.getCompleteorchestration()?.getDetails(), + action.getTerminateorchestration()?.getReason(), + action.getSendevent()?.getData(), + action.getSendentitymessage()?.getEntityoperationsignaled()?.getInput(), + action.getSendentitymessage()?.getEntityoperationcalled()?.getInput(), + ); + for (const event of action.getCompleteorchestration()?.getCarryovereventsList() ?? []) { + this._historyFields(event, fields); + } + for (const event of action.getRewindorchestration()?.getNewhistoryList() ?? []) { + this._historyFields(event, fields); + } + } + } else if (message instanceof pb.EntityBatchResult) { + fields.push(message.getEntitystate()); + for (const result of message.getResultsList()) fields.push(result.getSuccess()?.getResult()); + for (const action of message.getActionsList()) { + fields.push(action.getSendsignal()?.getInput(), action.getStartneworchestration()?.getInput()); + } + } else if (message instanceof pb.EntityBatchRequest) { + fields.push(message.getEntitystate()); + for (const operation of message.getOperationsList()) fields.push(operation.getInput()); + } else if (message instanceof pb.EntityRequest) { + fields.push(message.getEntitystate()); + } + const candidates: { field: StringValue; value: string; size: number }[] = []; + for (const field of fields) { + const value = field?.getValue(); + if (!field || !value) continue; + const size = Buffer.byteLength(value, "utf8"); + if (size < this._threshold) continue; + candidates.push({ field, value, size }); + } + try { + if (candidates.length) signal?.throwIfAborted(); + for (const { size } of candidates) { + if (size > this._maximum) { + throw new PayloadStorageException( + `Payload size ${size} bytes exceeds the configured maximum of ${this._maximum} bytes.`, + ); + } + } + await runBounded( + candidates.map(({ field, value }) => async () => { + field.setValue(await this._options.store.upload(value, signal)); + }), + signal, + ); + } catch (error) { + if ( + !isPermanentStorageFailure(error) || + !(message instanceof pb.ActivityResponse || message instanceof pb.OrchestratorResponse) + ) + throw error; + const details = newFailureDetails(error); + details.setIsnonretriable(true); + if (message instanceof pb.ActivityResponse) { + message.clearResult(); + message.setFailuredetails(details); + } else { + message.clearCustomstatus(); + message.setIspartial(false); + message.clearChunkindex(); + message.setActionsList([ + new pb.OrchestratorAction().setCompleteorchestration( + new pb.CompleteOrchestrationAction() + .setOrchestrationstatus(pb.OrchestrationStatus.ORCHESTRATION_STATUS_FAILED) + .setFailuredetails(details), + ), + ]); + } + } + } + + async resolve(message: unknown, signal?: AbortSignal): Promise { + const fields: (StringValue | undefined)[] = []; + this._responseFields(message, fields); + const operations: (() => Promise)[] = []; + for (const field of fields) { + const value = field?.getValue(); + if (field && value && this._options.store.isKnownPayloadToken(value)) { + operations.push(async () => { + field.setValue(await this._options.store.download(value, signal)); + }); + } + } + await runBounded(operations, signal); + } + + private _responseFields(message: unknown, fields: (StringValue | undefined)[]): void { + if (message instanceof pb.GetInstanceResponse) { + this._stateFields(message.getOrchestrationstate(), fields); + } else if (message instanceof pb.QueryInstancesResponse) { + for (const state of message.getOrchestrationstateList()) this._stateFields(state, fields); + } else if (message instanceof pb.HistoryChunk) { + for (const event of message.getEventsList()) this._historyFields(event, fields); + } else if (message instanceof pb.GetEntityResponse) { + fields.push(message.getEntity()?.getSerializedstate()); + } else if (message instanceof pb.QueryEntitiesResponse) { + for (const entity of message.getEntitiesList()) fields.push(entity.getSerializedstate()); + } else if (message instanceof pb.WorkItem) { + this._responseFields(message.getActivityrequest(), fields); + this._responseFields(message.getOrchestratorrequest(), fields); + this._responseFields(message.getEntityrequest(), fields); + this._responseFields(message.getEntityrequestv2(), fields); + } else if (message instanceof pb.ActivityRequest) { + fields.push(message.getInput()); + } else if (message instanceof pb.OrchestratorRequest) { + for (const event of [...message.getPasteventsList(), ...message.getNeweventsList()]) { + this._historyFields(event, fields); + } + } else if (message instanceof pb.EntityBatchRequest) { + fields.push(message.getEntitystate()); + for (const operation of message.getOperationsList()) fields.push(operation.getInput()); + } else if (message instanceof pb.EntityRequest) { + fields.push(message.getEntitystate()); + for (const event of message.getOperationrequestsList()) this._historyFields(event, fields); + } + } + + private _stateFields(state: pb.OrchestrationState | undefined, fields: (StringValue | undefined)[]): void { + if (state) fields.push(state.getInput(), state.getOutput(), state.getCustomstatus()); + } + + private _historyFields(event: pb.HistoryEvent, fields: (StringValue | undefined)[]): void { + fields.push( + event.getExecutionstarted()?.getInput(), + event.getExecutioncompleted()?.getResult(), + event.getExecutionterminated()?.getInput(), + event.getTaskscheduled()?.getInput(), + event.getTaskcompleted()?.getResult(), + event.getSuborchestrationinstancecreated()?.getInput(), + event.getSuborchestrationinstancecompleted()?.getResult(), + event.getEventsent()?.getInput(), + event.getEventraised()?.getInput(), + event.getGenericevent()?.getData(), + event.getContinueasnew()?.getInput(), + event.getExecutionsuspended()?.getInput(), + event.getExecutionresumed()?.getInput(), + event.getExecutionrewound()?.getReason(), + event.getEntityoperationsignaled()?.getInput(), + event.getEntityoperationcalled()?.getInput(), + event.getEntityoperationcompleted()?.getOutput(), + ); + this._stateFields(event.getHistorystate()?.getOrchestrationstate(), fields); + } +} + +export function isPermanentStorageFailure(error: unknown): error is Error { + if (error instanceof PayloadStorageException) return true; + if (!(error instanceof Error) || !("statusCode" in error)) return false; + const status = error.statusCode; + return typeof status === "number" && status >= 400 && status < 500 && status !== 408 && status !== 429; +} diff --git a/packages/durabletask-js/src/payloads/payload-store.ts b/packages/durabletask-js/src/payloads/payload-store.ts new file mode 100644 index 0000000..2eafae8 --- /dev/null +++ b/packages/durabletask-js/src/payloads/payload-store.ts @@ -0,0 +1,39 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +/** Confirmed results of a conditional, ownership-checked payload deletion. */ +export enum PayloadDeleteOutcome { + Unspecified = 0, + Deleted = 1, + AlreadyAbsent = 2, + NotStoreOwned = 3, +} + +/** Out-of-band storage for serialized UTF-8 payloads. Implementations must honor cancellation. */ +export abstract class PayloadStore { + abstract upload(payload: string, signal?: AbortSignal): Promise; + abstract download(token: string, signal?: AbortSignal): Promise; + abstract isKnownPayloadToken(value: string): boolean; + + /** Optional cleanup support. Never delete a blob without verifying ownership and its ETag. */ + async delete(_token: string, _signal?: AbortSignal): Promise { + throw new Error("This payload store does not support deletion."); + } +} + +/** A permanent payload storage or size-limit failure. */ +export class PayloadStorageException extends Error { + constructor(message: string, options?: { cause?: unknown }) { + super(message, options); + this.name = "PayloadStorageException"; + } +} + +/** Opt-in payload externalization. Configure the same store on all clients and workers. */ +export interface PayloadStorageOptions { + store: PayloadStore; + /** Inclusive serialized UTF-8 size threshold. Defaults to 256 KiB; maximum 1 MiB. */ + thresholdBytes?: number; + /** Maximum serialized size of one externalized payload. Defaults to 10 MiB. */ + maxPayloadBytes?: number; +} diff --git a/packages/durabletask-js/src/payloads/run-bounded.ts b/packages/durabletask-js/src/payloads/run-bounded.ts new file mode 100644 index 0000000..f6646f6 --- /dev/null +++ b/packages/durabletask-js/src/payloads/run-bounded.ts @@ -0,0 +1,32 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +/** Stops dispatch on failure/cancellation, drains started I/O, then throws the first message-order failure. */ +export async function runBounded(operations: (() => Promise)[], signal?: AbortSignal): Promise { + const active = new Set>(); + const failures = new Map(); + let cancellationPreventedDispatch = false; + for (let ordinal = 0; ordinal < operations.length; ordinal++) { + if (active.size >= 8) await Promise.race(active); + if (failures.size) break; + if (signal?.aborted) { + cancellationPreventedDispatch = true; + break; + } + const operation = operations[ordinal]; + const tracked = (async () => { + try { + await operation(); + } catch (error) { + failures.set(ordinal, error); + } + })(); + active.add(tracked); + void tracked.then(() => active.delete(tracked)); + // Observe synchronously rejected operations before claiming the next dispatch. + await Promise.resolve(); + } + await Promise.all(active); + if (failures.size) throw failures.get(Math.min(...failures.keys())); + if (cancellationPreventedDispatch) signal?.throwIfAborted(); +} diff --git a/packages/durabletask-js/test/payloads.spec.ts b/packages/durabletask-js/test/payloads.spec.ts new file mode 100644 index 0000000..5c466e9 --- /dev/null +++ b/packages/durabletask-js/test/payloads.spec.ts @@ -0,0 +1,786 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { StringValue, Int32Value } from "google-protobuf/google/protobuf/wrappers_pb"; +import * as pb from "../src/proto/orchestrator_service_pb"; +import { PayloadProcessor } from "../src/payloads/payload-processor"; +import { PayloadStore, PayloadStorageException } from "../src/payloads/payload-store"; +import { runBounded } from "../src/payloads/run-bounded"; +import { NoOpLogger } from "../src/types/logger.type"; +import { + newEventRaisedEvent, + newExecutionStartedEvent, + newOrchestratorStartedEvent, +} from "../src/utils/pb-helper.util"; +import { OrchestrationExecutor } from "../src/worker/orchestration-executor"; +import { Registry } from "../src/worker/registry"; + +class MemoryPayloadStore extends PayloadStore { + readonly values = new Map(); + uploads = 0; + async upload(value: string): Promise { + const token = `test:${++this.uploads}`; + this.values.set(token, value); + return token; + } + async download(token: string): Promise { + const value = this.values.get(token); + if (value === undefined) throw new Error("Missing payload"); + return value; + } + isKnownPayloadToken(value: string): boolean { + return value.startsWith("test:"); + } +} + +const text = (value: string) => new StringValue().setValue(value); +const MIB = 1024 * 1024; +const nextTurn = () => new Promise((resolve) => setImmediate(resolve)); + +function gate() { + let release!: () => void; + const promise = new Promise((resolve) => { + release = resolve; + }); + return { promise, release }; +} + +async function continueAsNewResponse(events: pb.HistoryEvent[]): Promise { + const registry = new Registry(); + registry.addNamedOrchestrator("Carryover", async function* (context) { + yield context.waitForExternalEvent("continue"); + context.continueAsNew("next", true, "next-version"); + }); + const result = await new OrchestrationExecutor(registry, new NoOpLogger()).execute( + "carryover-instance", + [], + [ + newOrchestratorStartedEvent(), + newExecutionStartedEvent("Carryover", "carryover-instance"), + ...events, + newEventRaisedEvent("continue", "null"), + ], + ); + return new pb.OrchestratorResponse() + .setInstanceid("carryover-instance") + .setCompletiontoken("completion") + .setActionsList(result.actions); +} + +describe("externalized payloads", () => { + let store: MemoryPayloadStore; + let processor: PayloadProcessor; + beforeEach(() => { + store = new MemoryPayloadStore(); + processor = new PayloadProcessor({ store }); + }); + + it.each([ + ["a".repeat(262143), false], + ["a".repeat(262144), true], + ["a".repeat(262145), true], + ["\u00e9".repeat(131072), true], + ["\ud83d\ude00".repeat(65536), true], + ["", false], + ])("uses the inclusive UTF-8 threshold (%#)", async (value, externalized) => { + const request = new pb.CreateInstanceRequest().setInput(text(value)); + await processor.externalize(request); + expect(store.uploads).toBe(externalized ? 1 : 0); + if (externalized) { + const response = new pb.GetInstanceResponse().setOrchestrationstate( + new pb.OrchestrationState().setInput(request.getInput()), + ); + await processor.resolve(response); + expect(response.getOrchestrationstate()!.getInput()!.getValue()).toBe(value); + } else { + expect(request.getInput()!.getValue()).toBe(value); + } + }); + + it("keeps absent payloads absent", async () => { + const request = new pb.CreateInstanceRequest(); + await processor.externalize(request); + expect(request.hasInput()).toBe(false); + expect(store.uploads).toBe(0); + }); + + it("maps large entity batches without the JavaScript variadic argument limit", async () => { + const request = new pb.EntityBatchRequest().setOperationsList( + Array.from({ length: 150000 }, () => new pb.OperationRequest().setInput(text("x"))), + ); + await expect(processor.resolve(request)).resolves.toBeUndefined(); + await expect(processor.externalize(request)).resolves.toBeUndefined(); + expect(store.uploads).toBe(0); + }); + + it("accepts exactly 10 MiB and rejects larger serialized payloads before storage", async () => { + const value = "x".repeat(10 * 1024 * 1024); + await processor.externalize(new pb.CreateInstanceRequest().setInput(text(value))); + await expect( + processor.externalize(new pb.CreateInstanceRequest().setInput(text(value + "x"))), + ).rejects.toBeInstanceOf(PayloadStorageException); + expect(store.uploads).toBe(1); + }); + + it.each([1048577, NaN, Infinity, -1, 0.5])("rejects invalid thresholds (%s)", (thresholdBytes) => { + expect(() => new PayloadProcessor({ store, thresholdBytes })).toThrow(RangeError); + }); + + it.each([ + () => new pb.CreateInstanceRequest().setInput(text("large")), + () => new pb.RaiseEventRequest().setInput(text("large")), + () => new pb.TerminateRequest().setOutput(text("large")), + () => new pb.SuspendRequest().setReason(text("large")), + () => new pb.ResumeRequest().setReason(text("large")), + () => new pb.RewindInstanceRequest().setReason(text("large")), + () => new pb.SignalEntityRequest().setInput(text("large")), + () => new pb.ActivityResponse().setResult(text("large")), + () => new pb.EntityRequest().setEntitystate(text("large")), + () => new pb.EntityBatchRequest().setEntitystate(text("large")), + ])("externalizes every client and worker scalar request (%#)", async (makeRequest) => { + processor = new PayloadProcessor({ store, thresholdBytes: 1 }); + await processor.externalize(makeRequest()); + expect(store.uploads).toBe(1); + }); + + it("externalizes all orchestration actions, custom status and rewritten history", async () => { + processor = new PayloadProcessor({ store, thresholdBytes: 1 }); + const response = new pb.OrchestratorResponse() + .setCustomstatus(text("status")) + .setActionsList([ + new pb.OrchestratorAction().setScheduletask(new pb.ScheduleTaskAction().setInput(text("activity"))), + new pb.OrchestratorAction().setCreatesuborchestration( + new pb.CreateSubOrchestrationAction().setInput(text("child")), + ), + new pb.OrchestratorAction().setCompleteorchestration( + new pb.CompleteOrchestrationAction().setResult(text("result")).setDetails(text("details")), + ), + new pb.OrchestratorAction().setTerminateorchestration( + new pb.TerminateOrchestrationAction().setReason(text("reason")), + ), + new pb.OrchestratorAction().setSendevent(new pb.SendEventAction().setData(text("event"))), + new pb.OrchestratorAction().setSendentitymessage( + new pb.SendEntityMessageAction().setEntityoperationcalled( + new pb.EntityOperationCalledEvent().setInput(text("entity")), + ), + ), + new pb.OrchestratorAction().setRewindorchestration( + new pb.RewindOrchestrationAction().setNewhistoryList([ + new pb.HistoryEvent().setExecutionstarted(new pb.ExecutionStartedEvent().setInput(text("rewritten"))), + ]), + ), + ]); + await processor.externalize(response); + expect([...store.values.values()]).toEqual([ + "status", + "activity", + "child", + "result", + "details", + "reason", + "event", + "entity", + "rewritten", + ]); + }); + + it("preserves entity result oneofs and externalizes state, outputs and outgoing operations", async () => { + processor = new PayloadProcessor({ store, thresholdBytes: 1 }); + const success = new pb.OperationResult().setSuccess(new pb.OperationResultSuccess().setResult(text("result"))); + const failure = new pb.OperationResult().setFailure(new pb.OperationResultFailure()); + const result = new pb.EntityBatchResult() + .setEntitystate(text("state")) + .setResultsList([success, failure]) + .setActionsList([ + new pb.OperationAction().setSendsignal(new pb.SendSignalAction().setInput(text("signal"))), + new pb.OperationAction().setStartneworchestration(new pb.StartNewOrchestrationAction().setInput(text("start"))), + ]); + await processor.externalize(result); + expect(store.uploads).toBe(4); + expect(success.hasSuccess()).toBe(true); + expect(failure.hasFailure()).toBe(true); + }); + + it("hydrates nested history state, activity and entity V1/V2 work items", async () => { + const token = await store.upload("resolved"); + const history = new pb.HistoryEvent().setHistorystate( + new pb.HistoryStateEvent().setOrchestrationstate( + new pb.OrchestrationState().setInput(text(token)).setOutput(text(token)).setCustomstatus(text(token)), + ), + ); + const messages = [ + new pb.WorkItem().setActivityrequest(new pb.ActivityRequest().setInput(text(token))), + new pb.WorkItem().setOrchestratorrequest(new pb.OrchestratorRequest().setPasteventsList([history])), + new pb.WorkItem().setEntityrequest( + new pb.EntityBatchRequest() + .setEntitystate(text(token)) + .setOperationsList([new pb.OperationRequest().setInput(text(token))]), + ), + new pb.WorkItem().setEntityrequestv2( + new pb.EntityRequest() + .setEntitystate(text(token)) + .setOperationrequestsList([ + new pb.HistoryEvent().setEntityoperationcalled(new pb.EntityOperationCalledEvent().setInput(text(token))), + ]), + ), + new pb.GetEntityResponse().setEntity(new pb.EntityMetadata().setSerializedstate(text(token))), + new pb.QueryEntitiesResponse().setEntitiesList([new pb.EntityMetadata().setSerializedstate(text(token))]), + new pb.QueryInstancesResponse().setOrchestrationstateList([new pb.OrchestrationState().setInput(text(token))]), + new pb.HistoryChunk().setEventsList([ + new pb.HistoryEvent().setExecutionrewound(new pb.ExecutionRewoundEvent().setReason(text(token))), + ]), + ]; + for (const message of messages) { + await processor.resolve(message); + expect(JSON.stringify(message.toObject())).not.toContain(token); + expect(JSON.stringify(message.toObject())).toContain("resolved"); + } + }); + + it.each([ + (value: StringValue) => new pb.HistoryEvent().setExecutionstarted(new pb.ExecutionStartedEvent().setInput(value)), + (value: StringValue) => + new pb.HistoryEvent().setExecutioncompleted(new pb.ExecutionCompletedEvent().setResult(value)), + (value: StringValue) => + new pb.HistoryEvent().setExecutionterminated(new pb.ExecutionTerminatedEvent().setInput(value)), + (value: StringValue) => new pb.HistoryEvent().setTaskscheduled(new pb.TaskScheduledEvent().setInput(value)), + (value: StringValue) => new pb.HistoryEvent().setTaskcompleted(new pb.TaskCompletedEvent().setResult(value)), + (value: StringValue) => + new pb.HistoryEvent().setSuborchestrationinstancecreated( + new pb.SubOrchestrationInstanceCreatedEvent().setInput(value), + ), + (value: StringValue) => + new pb.HistoryEvent().setSuborchestrationinstancecompleted( + new pb.SubOrchestrationInstanceCompletedEvent().setResult(value), + ), + (value: StringValue) => new pb.HistoryEvent().setEventsent(new pb.EventSentEvent().setInput(value)), + (value: StringValue) => new pb.HistoryEvent().setEventraised(new pb.EventRaisedEvent().setInput(value)), + (value: StringValue) => new pb.HistoryEvent().setGenericevent(new pb.GenericEvent().setData(value)), + (value: StringValue) => new pb.HistoryEvent().setContinueasnew(new pb.ContinueAsNewEvent().setInput(value)), + (value: StringValue) => + new pb.HistoryEvent().setExecutionsuspended(new pb.ExecutionSuspendedEvent().setInput(value)), + (value: StringValue) => new pb.HistoryEvent().setExecutionresumed(new pb.ExecutionResumedEvent().setInput(value)), + (value: StringValue) => new pb.HistoryEvent().setExecutionrewound(new pb.ExecutionRewoundEvent().setReason(value)), + (value: StringValue) => + new pb.HistoryEvent().setEntityoperationsignaled(new pb.EntityOperationSignaledEvent().setInput(value)), + (value: StringValue) => + new pb.HistoryEvent().setEntityoperationcalled(new pb.EntityOperationCalledEvent().setInput(value)), + (value: StringValue) => + new pb.HistoryEvent().setEntityoperationcompleted(new pb.EntityOperationCompletedEvent().setOutput(value)), + ])("hydrates and rewrites every payload-bearing history oneof (%#)", async (makeEvent) => { + processor = new PayloadProcessor({ store, thresholdBytes: 1 }); + const event = makeEvent(text(await store.upload("payload"))); + const originalCase = event.getEventtypeCase(); + await processor.resolve(new pb.HistoryChunk().setEventsList([event])); + expect(JSON.stringify(event.toObject())).toContain("payload"); + const response = new pb.OrchestratorResponse().setActionsList([ + new pb.OrchestratorAction().setRewindorchestration(new pb.RewindOrchestrationAction().setNewhistoryList([event])), + ]); + await processor.externalize(response); + expect([...store.values.values()]).toEqual(["payload", "payload"]); + expect(event.getEventtypeCase()).toBe(originalCase); + }); + + it("converts permanent activity upload failures without sending the result", async () => { + store.upload = async () => { + throw Object.assign(new Error("denied"), { statusCode: 403 }); + }; + processor = new PayloadProcessor({ store, thresholdBytes: 1 }); + const response = new pb.ActivityResponse().setResult(text("large")).setCompletiontoken("completion"); + await processor.externalize(response); + expect(response.hasResult()).toBe(false); + expect(response.getFailuredetails()!.getIsnonretriable()).toBe(true); + expect(response.getCompletiontoken()).toBe("completion"); + }); + + it("clears partial orchestration responses before a non-retriable failure", async () => { + processor = new PayloadProcessor({ store, thresholdBytes: 1, maxPayloadBytes: 2 }); + const response = new pb.OrchestratorResponse() + .setCustomstatus(text("large")) + .setIspartial(true) + .setChunkindex(new Int32Value().setValue(2)) + .setActionsList([new pb.OrchestratorAction().setScheduletask(new pb.ScheduleTaskAction().setInput(text("x")))]); + await processor.externalize(response); + expect(response.hasCustomstatus()).toBe(false); + expect(response.getIspartial()).toBe(false); + expect(response.hasChunkindex()).toBe(false); + expect(response.getActionsList()).toHaveLength(1); + const complete = response.getActionsList()[0].getCompleteorchestration()!; + expect(complete.getOrchestrationstatus()).toBe(pb.OrchestrationStatus.ORCHESTRATION_STATUS_FAILED); + expect(complete.getFailuredetails()!.getIsnonretriable()).toBe(true); + }); + + it.each([408, 429, 500, 503])("propagates transient upload failure %s", async (statusCode) => { + const error = Object.assign(new Error("transient"), { statusCode }); + store.upload = async () => { + throw error; + }; + processor = new PayloadProcessor({ store, thresholdBytes: 1 }); + await expect(processor.externalize(new pb.ActivityResponse().setResult(text("large")))).rejects.toBe(error); + }); + + describe("continue-as-new carryover", () => { + it("externalizes five 1 MiB runtime events below the 4 MiB wire limit and preserves hydrated history", async () => { + const values = Array.from({ length: 5 }, (_, index) => `${index}${"x".repeat(MIB - 3)}`); + const response = await continueAsNewResponse( + values.map((value, index) => newEventRaisedEvent(index < 2 ? "FIRST" : "SECOND", JSON.stringify(value))), + ); + const original = response.toObject(); + const action = response.getActionsList()[0]; + const completion = action.getCompleteorchestration()!; + const events = completion.getCarryovereventsList(); + expect(action.getId()).toBe(1); + expect(completion.getOrchestrationstatus()).toBe(pb.OrchestrationStatus.ORCHESTRATION_STATUS_CONTINUED_AS_NEW); + expect(events.map((event) => event.getEventraised()!.getName())).toEqual([ + "first", + "first", + "second", + "second", + "second", + ]); + expect(events.map((event) => Buffer.byteLength(event.getEventraised()!.getInput()!.getValue(), "utf8"))).toEqual( + Array(5).fill(MIB), + ); + expect(response.serializeBinary().byteLength).toBeGreaterThan(4 * MIB); + + await processor.externalize(response); + + const binary = response.serializeBinary(); + expect(binary.byteLength).toBeLessThan(4 * MIB); + expect(store.uploads).toBe(5); + expect([...store.values.values()]).toEqual(values.map((value) => JSON.stringify(value))); + const wireResponse = pb.OrchestratorResponse.deserializeBinary(binary); + const carryover = wireResponse.getActionsList()[0].getCompleteorchestration()!.getCarryovereventsList(); + expect( + carryover.every((event) => store.isKnownPayloadToken(event.getEventraised()!.getInput()!.getValue())), + ).toBe(true); + + await processor.resolve(new pb.HistoryChunk().setEventsList(carryover)); + + expect(carryover.map((event) => JSON.parse(event.getEventraised()!.getInput()!.getValue()))).toEqual(values); + expect(wireResponse.toObject()).toEqual(original); + }); + + it.each([ + ["below threshold", "\u00e9".repeat(131070) + "a", 262143, false], + ["at threshold", "\u00e9".repeat(131071), 262144, true], + ["above threshold", "\u00e9".repeat(131071) + "a", 262145, true], + ["below maximum", "\u00e9".repeat(524286) + "a", MIB - 1, true], + ["at maximum", "\u00e9".repeat(524287), MIB, true], + ])("applies serialized UTF-8 carryover boundaries %s", async (_label, value, size, externalized) => { + processor = new PayloadProcessor({ store, maxPayloadBytes: MIB }); + const encoded = JSON.stringify(value); + expect(Buffer.byteLength(encoded, "utf8")).toBe(size); + const response = await continueAsNewResponse([newEventRaisedEvent("buffered", encoded)]); + + await processor.externalize(response); + + const completion = response.getActionsList()[0].getCompleteorchestration()!; + const input = completion.getCarryovereventsList()[0].getEventraised()!.getInput()!.getValue(); + expect(completion.getOrchestrationstatus()).toBe(pb.OrchestrationStatus.ORCHESTRATION_STATUS_CONTINUED_AS_NEW); + expect(store.uploads).toBe(externalized ? 1 : 0); + expect(store.isKnownPayloadToken(input)).toBe(externalized); + await processor.resolve(new pb.HistoryChunk().setEventsList(completion.getCarryovereventsList())); + expect(completion.getCarryovereventsList()[0].getEventraised()!.getInput()!.getValue()).toBe(encoded); + }); + + it.each([ + ["configured maximum", "\u00e9".repeat(524287) + "a", MIB, MIB + 1], + ["default maximum", "x".repeat(10 * MIB - 1), undefined, 10 * MIB + 1], + ])("converts carryover above the %s to a non-retriable failure", async (_label, value, maximum, size) => { + processor = new PayloadProcessor({ store, maxPayloadBytes: maximum }); + const encoded = JSON.stringify(value); + expect(Buffer.byteLength(encoded, "utf8")).toBe(size); + const response = await continueAsNewResponse([newEventRaisedEvent("buffered", encoded)]); + + await processor.externalize(response); + + expect(store.uploads).toBe(0); + expect(store.values.size).toBe(0); + expect(response.getActionsList()).toHaveLength(1); + const completion = response.getActionsList()[0].getCompleteorchestration()!; + expect(completion.getOrchestrationstatus()).toBe(pb.OrchestrationStatus.ORCHESTRATION_STATUS_FAILED); + expect(completion.getCarryovereventsList()).toEqual([]); + expect(completion.getFailuredetails()!.getErrortype()).toBe("PayloadStorageException"); + expect(completion.getFailuredetails()!.getIsnonretriable()).toBe(true); + expect(completion.getFailuredetails()!.getErrormessage()).toBe( + `Payload size ${size} bytes exceeds the configured maximum of ${maximum ?? 10 * MIB} bytes.`, + ); + expect(response.getInstanceid()).toBe("carryover-instance"); + expect(response.getCompletiontoken()).toBe("completion"); + }); + }); + + describe("whole-message size preflight", () => { + it("fails an oversized action before uploading or mutating a valid custom status", async () => { + processor = new PayloadProcessor({ store, maxPayloadBytes: MIB }); + const status = text("s".repeat(262144)); + const input = text("x".repeat(MIB + 1)); + const response = new pb.OrchestratorResponse() + .setInstanceid("instance") + .setCompletiontoken("completion") + .setCustomstatus(status) + .setIspartial(true) + .setChunkindex(new Int32Value().setValue(7)) + .setActionsList([new pb.OrchestratorAction().setScheduletask(new pb.ScheduleTaskAction().setInput(input))]); + + await processor.externalize(response); + + expect(store.uploads).toBe(0); + expect(store.values.size).toBe(0); + expect(status.getValue()).toBe("s".repeat(262144)); + expect(input.getValue()).toBe("x".repeat(MIB + 1)); + expect(response.hasCustomstatus()).toBe(false); + expect(response.getIspartial()).toBe(false); + expect(response.hasChunkindex()).toBe(false); + expect(response.getActionsList()).toHaveLength(1); + const completion = response.getActionsList()[0].getCompleteorchestration()!; + expect(completion.getOrchestrationstatus()).toBe(pb.OrchestrationStatus.ORCHESTRATION_STATUS_FAILED); + expect(completion.getFailuredetails()!.getErrortype()).toBe("PayloadStorageException"); + expect(completion.getFailuredetails()!.getIsnonretriable()).toBe(true); + expect(completion.getFailuredetails()!.getErrormessage()).toBe( + `Payload size ${MIB + 1} bytes exceeds the configured maximum of ${MIB} bytes.`, + ); + expect(response.getInstanceid()).toBe("instance"); + expect(response.getCompletiontoken()).toBe("completion"); + }); + + it("rejects an oversized entity operation before uploading or mutating valid state", async () => { + processor = new PayloadProcessor({ store, maxPayloadBytes: MIB }); + const request = new pb.EntityBatchRequest() + .setEntitystate(text("s".repeat(262144))) + .setOperationsList([new pb.OperationRequest().setInput(text("x".repeat(MIB + 1)))]); + const original = request.toObject(); + + await expect(processor.externalize(request)).rejects.toBeInstanceOf(PayloadStorageException); + + expect(store.uploads).toBe(0); + expect(store.values.size).toBe(0); + expect(request.toObject()).toEqual(original); + }); + + it("rejects oversized client input without storage or field mutations", async () => { + processor = new PayloadProcessor({ store, maxPayloadBytes: MIB }); + const request = new pb.CreateInstanceRequest().setInput(text("x".repeat(MIB + 1))); + const original = request.toObject(); + + await expect(processor.externalize(request)).rejects.toBeInstanceOf(PayloadStorageException); + + expect(store.uploads).toBe(0); + expect(store.values.size).toBe(0); + expect(request.toObject()).toEqual(original); + }); + + it("converts oversized activity output while preserving completion metadata", async () => { + processor = new PayloadProcessor({ store, maxPayloadBytes: MIB }); + const result = text("x".repeat(MIB + 1)); + const response = new pb.ActivityResponse() + .setInstanceid("instance") + .setTaskid(17) + .setCompletiontoken("completion") + .setResult(result); + + await processor.externalize(response); + + expect(store.uploads).toBe(0); + expect(store.values.size).toBe(0); + expect(result.getValue()).toBe("x".repeat(MIB + 1)); + expect(response.hasResult()).toBe(false); + expect(response.getFailuredetails()!.getErrortype()).toBe("PayloadStorageException"); + expect(response.getFailuredetails()!.getIsnonretriable()).toBe(true); + expect(response.getFailuredetails()!.getErrormessage()).toBe( + `Payload size ${MIB + 1} bytes exceeds the configured maximum of ${MIB} bytes.`, + ); + expect(response.getInstanceid()).toBe("instance"); + expect(response.getTaskid()).toBe(17); + expect(response.getCompletiontoken()).toBe("completion"); + }); + + it("detects invalid size before a hypothetical earlier transient upload failure", async () => { + processor = new PayloadProcessor({ store, maxPayloadBytes: MIB }); + const upload = jest + .spyOn(store, "upload") + .mockRejectedValue(Object.assign(new Error("offline"), { statusCode: 503 })); + const response = new pb.OrchestratorResponse() + .setCustomstatus(text("s".repeat(262144))) + .setActionsList([ + new pb.OrchestratorAction().setScheduletask(new pb.ScheduleTaskAction().setInput(text("x".repeat(MIB + 1)))), + ]); + + await expect(processor.externalize(response)).resolves.toBeUndefined(); + + expect(upload).not.toHaveBeenCalled(); + expect(store.values.size).toBe(0); + const completion = response.getActionsList()[0].getCompleteorchestration()!; + expect(completion.getOrchestrationstatus()).toBe(pb.OrchestrationStatus.ORCHESTRATION_STATUS_FAILED); + expect(completion.getFailuredetails()!.getErrortype()).toBe("PayloadStorageException"); + expect(completion.getFailuredetails()!.getIsnonretriable()).toBe(true); + }); + + it.each([ + () => new pb.CreateInstanceRequest().setInput(text("x".repeat(MIB + 1))), + () => new pb.ActivityResponse().setResult(text("x".repeat(MIB + 1))), + () => new pb.EntityBatchRequest().setEntitystate(text("x".repeat(MIB + 1))), + () => + new pb.OrchestratorResponse() + .setCustomstatus(text("s".repeat(262144))) + .setIspartial(true) + .setChunkindex(new Int32Value().setValue(7)) + .setActionsList([ + new pb.OrchestratorAction().setScheduletask( + new pb.ScheduleTaskAction().setInput(text("x".repeat(MIB + 1))), + ), + ]), + ])("honors precancellation before deterministic size failure (%#)", async (makeMessage) => { + processor = new PayloadProcessor({ store, maxPayloadBytes: MIB }); + const abort = new AbortController(); + abort.abort(); + const message = makeMessage(); + const original = message.toObject(); + + await expect(processor.externalize(message, abort.signal)).rejects.toBe(abort.signal.reason); + + expect(abort.signal.reason.name).toBe("AbortError"); + expect(store.uploads).toBe(0); + expect(store.values.size).toBe(0); + expect(message.toObject()).toEqual(original); + }); + + it.each([ + () => new pb.CreateInstanceRequest(), + () => new pb.CreateInstanceRequest().setInput(text("")), + () => new pb.CreateInstanceRequest().setInput(text("abc")), + ])( + "keeps canceled no-op messages unchanged even when the maximum is below the threshold (%#)", + async (makeMessage) => { + processor = new PayloadProcessor({ store, thresholdBytes: 4, maxPayloadBytes: 2 }); + const abort = new AbortController(); + abort.abort(); + const message = makeMessage(); + const original = message.toObject(); + + await expect(processor.externalize(message, abort.signal)).resolves.toBeUndefined(); + + expect(store.uploads).toBe(0); + expect(message.toObject()).toEqual(original); + }, + ); + + it("externalizes every eligible valid field and leaves inline, empty, and absent values untouched", async () => { + processor = new PayloadProcessor({ store, thresholdBytes: 4, maxPayloadBytes: 4 }); + const request = new pb.EntityBatchRequest() + .setEntitystate(text("1234")) + .setOperationsList([ + new pb.OperationRequest().setInput(text("\u00e9\u00e9")), + new pb.OperationRequest().setInput(text("abc")), + new pb.OperationRequest().setInput(text("")), + new pb.OperationRequest(), + ]); + const original = request.toObject(); + + await processor.externalize(request); + + expect([...store.values.values()]).toEqual(["1234", "\u00e9\u00e9"]); + expect(request.getEntitystate()!.getValue()).toBe("test:1"); + expect(request.getOperationsList()[0].getInput()!.getValue()).toBe("test:2"); + expect(request.getOperationsList()[1].getInput()!.getValue()).toBe("abc"); + expect(request.getOperationsList()[2].getInput()!.getValue()).toBe(""); + expect(request.getOperationsList()[3].hasInput()).toBe(false); + await processor.resolve(request); + expect(request.toObject()).toEqual(original); + }); + }); + + describe("in-flight payload I/O", () => { + it.each([false, true])( + "classifies the lowest-message-order failure after drainage (permanent: %s)", + async (permanent) => { + processor = new PayloadProcessor({ store, thresholdBytes: 1 }); + const earlier = gate(); + const later = gate(); + const bothStarted = gate(); + const laterFinished = gate(); + const firstError = Object.assign(new Error("earlier failure"), { statusCode: permanent ? 403 : 503 }); + const secondError = permanent + ? Object.assign(new Error("later transient failure"), { statusCode: 503 }) + : new PayloadStorageException("later permanent failure"); + const started: string[] = []; + const finished: string[] = []; + let active = 0; + let settled = false; + store.upload = async (value) => { + started.push(value); + active++; + if (started.length === 2) bothStarted.release(); + try { + await (value === "earlier" ? earlier.promise : later.promise); + throw value === "earlier" ? firstError : secondError; + } finally { + finished.push(value); + active--; + if (value === "later") laterFinished.release(); + } + }; + const response = new pb.OrchestratorResponse() + .setCustomstatus(text("earlier")) + .setIspartial(true) + .setChunkindex(new Int32Value().setValue(2)) + .setActionsList([ + new pb.OrchestratorAction().setScheduletask(new pb.ScheduleTaskAction().setInput(text("later"))), + ]); + const original = response.toObject(); + const pending = processor.externalize(response).then( + () => { + settled = true; + }, + (error: unknown) => { + settled = true; + return error; + }, + ); + await bothStarted.promise; + later.release(); + await laterFinished.promise; + await nextTurn(); + const beforeDrain = { settled, active, finished: [...finished] }; + earlier.release(); + const error = await pending; + + expect(beforeDrain).toEqual({ settled: false, active: 1, finished: ["later"] }); + expect(started).toEqual(["earlier", "later"]); + expect(finished).toEqual(["later", "earlier"]); + expect(active).toBe(0); + if (permanent) { + expect(error).toBeUndefined(); + expect(response.hasCustomstatus()).toBe(false); + expect(response.getIspartial()).toBe(false); + expect(response.hasChunkindex()).toBe(false); + expect(response.getActionsList()).toHaveLength(1); + const completion = response.getActionsList()[0].getCompleteorchestration()!; + expect(completion.getOrchestrationstatus()).toBe(pb.OrchestrationStatus.ORCHESTRATION_STATUS_FAILED); + expect(completion.getFailuredetails()!.getErrormessage()).toBe("earlier failure"); + expect(completion.getFailuredetails()!.getIsnonretriable()).toBe(true); + } else { + expect(error).toBe(firstError); + expect(response.toObject()).toEqual(original); + } + const atSettlement = response.toObject(); + await nextTurn(); + expect(response.toObject()).toEqual(atSettlement); + }, + ); + + it.each(["externalize", "resolve"] as const)( + "drains started %s mutations before cancellation settles", + async (mode) => { + processor = new PayloadProcessor({ store, thresholdBytes: 1 }); + const abort = new AbortController(); + const started = gate(); + const hold = gate(); + const first = mode === "externalize" ? "first" : "test:first"; + const second = mode === "externalize" ? "second" : "test:second"; + store.values.set("test:first", "resolved-first"); + store.values.set("test:second", "resolved-second"); + const upload = store.upload.bind(store); + const download = store.download.bind(store); + const calls: string[] = []; + let active = 0; + let settled = false; + const io = async (value: string) => { + calls.push(value); + active++; + abort.abort(); + started.release(); + try { + await hold.promise; + return mode === "externalize" ? await upload(value) : await download(value); + } finally { + active--; + } + }; + if (mode === "externalize") store.upload = io; + else store.download = io; + const request = new pb.EntityBatchRequest() + .setEntitystate(text(first)) + .setOperationsList([new pb.OperationRequest().setInput(text(second))]); + const pending = processor[mode](request, abort.signal).then( + () => { + settled = true; + }, + (error: unknown) => { + settled = true; + return error; + }, + ); + await started.promise; + await nextTurn(); + const beforeDrain = { settled, active, state: request.getEntitystate()!.getValue() }; + hold.release(); + const error = await pending; + + expect(beforeDrain).toEqual({ settled: false, active: 1, state: first }); + expect(calls).toEqual([first]); + expect(error).toBe(abort.signal.reason); + expect(active).toBe(0); + expect(request.getEntitystate()!.getValue()).toBe(mode === "externalize" ? "test:1" : "resolved-first"); + expect(request.getOperationsList()[0].getInput()!.getValue()).toBe(second); + const atSettlement = request.toObject(); + await nextTurn(); + expect(request.toObject()).toEqual(atSettlement); + }, + ); + }); +}); + +describe("bounded payload I/O", () => { + it("caps concurrency at eight", async () => { + let active = 0; + let maximum = 0; + await runBounded( + Array.from({ length: 40 }, () => async () => { + maximum = Math.max(maximum, ++active); + await new Promise((resolve) => setImmediate(resolve)); + active--; + }), + ); + expect(maximum).toBe(8); + expect(active).toBe(0); + }); + + it("stops dispatch, drains every started operation and throws the lowest-ordinal failure", async () => { + const first = new Error("first in message order"); + const later = new Error("first to finish"); + const started: number[] = []; + const completed: number[] = []; + await expect( + runBounded( + Array.from({ length: 20 }, (_, index) => async () => { + started.push(index); + if (index === 1) throw later; + await new Promise((resolve) => setImmediate(resolve)); + completed.push(index); + if (index === 0) throw first; + }), + ), + ).rejects.toBe(first); + expect(started).toHaveLength(2); + expect(completed).toEqual([0]); + }); + + it("drains on cancellation and does not observe cancellation for no-op messages", async () => { + const abort = new AbortController(); + let completed = 0; + await expect( + runBounded( + Array.from({ length: 20 }, () => async () => { + abort.abort(); + await new Promise((resolve) => setImmediate(resolve)); + completed++; + }), + abort.signal, + ), + ).rejects.toBe(abort.signal.reason); + expect(completed).toBe(1); + await expect(runBounded([], abort.signal)).resolves.toBeUndefined(); + }); +});