feat: [4/6] add payload purge protocol and client lifecycle

CHANGELOG.md

@@ -2,6 +2,9 @@
 
 ### New
 
+- Add the DTS payload-purge protocol and `setLargePayloadAutoPurge()` client lifecycle,
+  including setting-first enable/disable and reserved-instance identity verification.
+  A compatible .NET worker or the later cleanup-worker phase is required; do not publish partial phases.
 - 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.
@@ -47,6 +50,10 @@
 
 ### Fixes
 
+- Deduplicate auto-purge runner startup only after an actual scheduler RPC duplicate response.
+  Preserve upload/metadata error identity, including matching semantic error types, and cancellation.
+- Determine single-instance purge routing from the instance ID regardless of payload configuration,
+  matching .NET management clients: ordinary IDs are orchestrations and `@name@key` IDs are entities.
 - 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.

README.md

@@ -48,6 +48,22 @@ Azure-managed builders expose `.useExternalizedPayloads({ store })` and require
 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.
 
+### Payload cleanup client protocol
+
+`client.setLargePayloadAutoPurge(enabled, batchSize?, signal?)` manages the authenticated DTS
+task-hub cleanup setting. Enabling writes the setting first, starts/deduplicates and verifies
+the reserved Running orchestration, then signals its batch size. Disabling only writes the
+backend setting. These steps are not transactional; earlier successful steps are not rolled back.
+
+This client operation does not require blob credentials. It requires a backend implementing
+the DTS `LargePayloadPurge` service and a compatible worker (the existing .NET implementation
+or the later phase 5 JavaScript runner). This phase does not register worker cleanup tasks.
+Do not publish the feature until all six phases are complete.
+
+Single-instance purges now identify the instance kind even on management-only clients without
+payload storage: ordinary IDs set `isOrchestration=true`, while `@name@key` entity IDs set it false.
+The existing timeout and completion-status behavior is unchanged.
+
 ## Long durable timers
 
 `createTimer(Date | seconds)` has no SDK-imposed total-duration cap. Like the Python SDK,

internal/protocol/protos/large_payload_purge.proto

@@ -0,0 +1,56 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+//
+// DTS-only contract from durabletask-dotnet 91739687ff647699079cdadd627daa00562982b7.
+
+syntax = "proto3";
+
+package microsoft.durabletask.largepayloads;
+
+option csharp_namespace = "Microsoft.DurableTask.Protobuf.LargePayloads";
+
+// The backend owns the tombstone ledger, task-hub setting, and retry scheduling.
+// The credentialed worker performs ownership-checked blob deletes and reports outcomes.
+service LargePayloadPurge {
+  rpc SetLargePayloadAutoPurge(SetLargePayloadAutoPurgeRequest) returns (SetLargePayloadAutoPurgeResponse);
+  rpc GetLargePayloadTombstones(GetLargePayloadTombstonesRequest) returns (GetLargePayloadTombstonesResponse);
+  rpc ReportLargePayloadPurgeResults(ReportLargePayloadPurgeResultsRequest) returns (ReportLargePayloadPurgeResultsResponse);
+}
+
+message SetLargePayloadAutoPurgeRequest {
+  bool enabled = 1;
+}
+
+message SetLargePayloadAutoPurgeResponse {}
+
+message LargePayloadTombstone {
+  // Opaque correlation token. Echo unchanged; never parse or reconstruct it.
+  string tombstone_token = 1;
+  string payload_token = 2;
+}
+
+enum LargePayloadPurgeDisposition {
+  LARGE_PAYLOAD_PURGE_DISPOSITION_UNSPECIFIED = 0;
+  LARGE_PAYLOAD_PURGE_DISPOSITION_DELETED = 1;
+  LARGE_PAYLOAD_PURGE_DISPOSITION_RETRY = 2;
+  LARGE_PAYLOAD_PURGE_DISPOSITION_QUARANTINED = 3;
+}
+
+message LargePayloadPurgeResult {
+  string tombstone_token = 1;
+  LargePayloadPurgeDisposition disposition = 2;
+}
+
+message GetLargePayloadTombstonesRequest {
+  int32 limit = 1;
+}
+
+message GetLargePayloadTombstonesResponse {
+  repeated LargePayloadTombstone tombstones = 1;
+}
+
+message ReportLargePayloadPurgeResultsRequest {
+  repeated LargePayloadPurgeResult results = 1;
+}
+
+message ReportLargePayloadPurgeResultsResponse {}

packages/durabletask-js/src/client/client.ts

@@ -40,6 +40,10 @@ 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 * as lp from "../proto/large_payload_purge_pb";
+import { LargePayloadPurgeService } from "../proto/large_payload_purge_grpc_pb";
+import { callPayloadPurge } from "../payloads/purge-transport";
+import { BlobPurgeConstants, validatePurgeBatchSize } from "../payloads/purge";
 import {
   CleanEntityStorageRequest,
   CleanEntityStorageResult,
@@ -207,6 +211,87 @@ export class TaskHubGrpcClient {
     }
   }
 
+  /**
+   * Enables or pauses DTS large-payload cleanup. Enabling starts/verifies the reserved singleton
+   * and enqueues its batch size (1..1000, default 500). Disabling only writes the backend setting.
+   * Steps are not transactional; errors and cancellation do not roll back an earlier successful step.
+   * A payload-enabled worker is required. Legacy v1 references are not deleted.
+   */
+  async setLargePayloadAutoPurge(enabled: boolean, batchSize = 500, signal?: AbortSignal): Promise<void> {
+    if (enabled) validatePurgeBatchSize(batchSize);
+    const cancellation = signal
+      ? AbortSignal.any([signal, this._payloadCancellation.signal])
+      : this._payloadCancellation.signal;
+    await callPayloadPurge(
+      this._stub,
+      LargePayloadPurgeService.setLargePayloadAutoPurge,
+      new lp.SetLargePayloadAutoPurgeRequest().setEnabled(enabled),
+      this._metadataGenerator,
+      cancellation,
+    );
+    if (!enabled) return;
+    const request = new pb.CreateInstanceRequest()
+      .setInstanceid(BlobPurgeConstants.instanceId)
+      .setName(BlobPurgeConstants.orchestrator)
+      .setVersion(new StringValue().setValue(""))
+      .setInput(new StringValue().setValue(JSON.stringify({ PurgeBatchSize: batchSize, PurgedCount: 0 })))
+      .setOrchestrationidreusepolicy(
+        new pb.OrchestrationIdReusePolicy().setReplaceablestatusList([
+          pb.OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED,
+          pb.OrchestrationStatus.ORCHESTRATION_STATUS_FAILED,
+          pb.OrchestrationStatus.ORCHESTRATION_STATUS_CANCELED,
+          pb.OrchestrationStatus.ORCHESTRATION_STATUS_TERMINATED,
+        ]),
+      );
+    let runnerAlreadyExists = false;
+    try {
+      const response = await this._call<pb.CreateInstanceRequest, pb.CreateInstanceResponse>(
+        this._stub.startInstance.bind(this._stub),
+        request,
+        this._metadataGenerator,
+        cancellation,
+        (error) => {
+          if (error instanceof Error && "code" in error && error.code === grpc.status.ALREADY_EXISTS) {
+            runnerAlreadyExists = true;
+          }
+          return false;
+        },
+      );
+      if (response.getInstanceid() !== BlobPurgeConstants.instanceId) {
+        throw new Error("The backend returned an unexpected auto-purge instance ID.");
+      }
+    } catch (error) {
+      cancellation.throwIfAborted();
+      if (!runnerAlreadyExists) throw error;
+    }
+    const started = await this._call<pb.GetInstanceRequest, pb.GetInstanceResponse>(
+      this._stub.waitForInstanceStart.bind(this._stub),
+      new pb.GetInstanceRequest().setInstanceid(BlobPurgeConstants.instanceId).setGetinputsandoutputs(false),
+      this._metadataGenerator,
+      cancellation,
+    );
+    const state = started.getOrchestrationstate();
+    if (
+      !started.getExists() ||
+      state?.getInstanceid() !== BlobPurgeConstants.instanceId ||
+      state?.getName() !== BlobPurgeConstants.orchestrator
+    ) {
+      throw new Error("The reserved auto-purge instance ID belongs to a different orchestration.");
+    }
+    if (state.getOrchestrationstatus() !== pb.OrchestrationStatus.ORCHESTRATION_STATUS_RUNNING) {
+      throw new Error("Auto-purge did not start in Running status.");
+    }
+    await this._call(
+      this._stub.raiseEvent.bind(this._stub),
+      new pb.RaiseEventRequest()
+        .setInstanceid(BlobPurgeConstants.instanceId)
+        .setName(BlobPurgeConstants.configurationEvent)
+        .setInput(new StringValue().setValue(JSON.stringify(batchSize))),
+      this._metadataGenerator,
+      cancellation,
+    );
+  }
+
   /**
    * Schedules a new orchestrator using the DurableTask client.
    *
@@ -815,6 +900,7 @@ export class TaskHubGrpcClient {
       const req = new pb.PurgeInstancesRequest();
       req.setInstanceid(instanceId);
       req.setRecursive(options?.recursive ?? false);
+      req.setIsorchestration(!instanceId.startsWith("@"));
 
       ClientLogs.purgingInstanceMetadata(this._logger, instanceId);
 

packages/durabletask-js/src/index.ts

@@ -8,6 +8,7 @@ export {
   PayloadStorageException,
   PayloadDeleteOutcome,
 } from "./payloads/payload-store";
+export { LargePayloadPurgeDisposition, LargePayloadTombstone, LargePayloadPurgeResult } from "./payloads/purge";
 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/purge-transport.ts

@@ -0,0 +1,33 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+import * as grpc from "@grpc/grpc-js";
+import { callWithMetadata, MetadataGenerator } from "../utils/grpc-helper.util";
+
+/** Reuses the effective sidecar's authenticated channel, including after worker channel recreation. */
+export async function callPayloadPurge<TReq, TRes>(
+  client: grpc.Client,
+  definition: grpc.MethodDefinition<TReq, TRes>,
+  request: TReq,
+  metadataGenerator?: MetadataGenerator,
+  signal?: AbortSignal,
+  deadline?: Date,
+): Promise<TRes> {
+  const response = await callWithMetadata<TReq, TRes | undefined>(
+    (req, metadata, callback) =>
+      client.makeUnaryRequest(
+        definition.path,
+        definition.requestSerialize,
+        definition.responseDeserialize,
+        req,
+        metadata,
+        { deadline },
+        callback,
+      ),
+    request,
+    metadataGenerator,
+    signal,
+  );
+  if (response === undefined) throw new Error("The large-payload purge RPC returned no response.");
+  return response;
+}

packages/durabletask-js/src/payloads/purge.ts

@@ -0,0 +1,41 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+export const BlobPurgeConstants = {
+  orchestrator: "BlobPurgeJobOrchestrator",
+  instanceId: "BlobPurgeJob-__dt_blob_payload_autopurge__",
+  fetch: "GetLargePayloadTombstonesActivity",
+  delete: "DeleteExternalBlobActivity",
+  report: "ReportLargePayloadPurgeResultsActivity",
+  configurationEvent: "SetBatchSize",
+} as const;
+
+export enum LargePayloadPurgeDisposition {
+  Unspecified = 0,
+  Deleted = 1,
+  Retry = 2,
+  Quarantined = 3,
+}
+
+// These activity/orchestration wire shapes intentionally retain .NET's default JSON property casing.
+export interface LargePayloadTombstone {
+  TombstoneToken: string;
+  PayloadToken: string;
+}
+export interface LargePayloadPurgeResult {
+  TombstoneToken: string;
+  Disposition: LargePayloadPurgeDisposition;
+}
+export interface BlobPurgeOutcome {
+  Disposition: LargePayloadPurgeDisposition;
+}
+export interface BlobPurgeJobRunRequest {
+  PurgeBatchSize: number;
+  PurgedCount: number;
+}
+
+export function validatePurgeBatchSize(batchSize: number): void {
+  if (!Number.isInteger(batchSize) || batchSize < 1 || batchSize > 1000) {
+    throw new RangeError("Purge batch size must be an integer between 1 and 1000.");
+  }
+}