feat: [2/6] integrate payloads with clients workers and history

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

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,

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,

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

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,
+    });
   }
 }
 

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,
     });

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<string> {
+    return "external:payload";
+  }
+  async download(): Promise<string> {
+    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"]);
+  });
+});