feat: [1/6] add core payload contracts and bounded processing

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,19 @@
 
 ### 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.
+- Skip store-recognized payload tokens before externalization size checks and uploads, so retries
+  of already or partially externalized messages hydrate back to the original data in one pass.
+- Preserve canceled payload processing as an error instead of a permanent worker failure, even
+  when the abort reason resembles a storage failure. Rethrow the drained runner's original error
+  without replacing earlier I/O failures with a later cancellation reason.
+- Include rewind-history input in payload hydration, externalization, and UTF-8 size preflight,
+  preserving the rewind reason and metadata.
 - 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.

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"

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",

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";

packages/durabletask-js/src/payloads/payload-processor.ts

@@ -0,0 +1,207 @@
+// 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<void> {
+    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 || this._options.store.isKnownPayloadToken(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 (
+        signal?.aborted ||
+        !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<void> {
+    const fields: (StringValue | undefined)[] = [];
+    this._responseFields(message, fields);
+    const operations: (() => Promise<void>)[] = [];
+    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.getExecutionrewound()?.getInput(),
+      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;
+}

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<string>;
+  abstract download(token: string, signal?: AbortSignal): Promise<string>;
+  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<PayloadDeleteOutcome> {
+    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;
+}

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<void>)[], signal?: AbortSignal): Promise<void> {
+  const active = new Set<Promise<void>>();
+  const failures = new Map<number, unknown>();
+  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();
+}