feat: [5/6] add reserved payload cleanup worker tasks

CHANGELOG.md

@@ -2,6 +2,9 @@
 
 ### New
 
+- Add reserved cleanup worker tasks with bounded deletion chunks, retry/quarantine handling,
+  configuration-preserving continue-as-new, authenticated current-channel RPCs and filter/version isolation.
+  This completes the cleanup runtime; aggregate release/consumer validation follows in phase 6.
 - 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.

README.md

@@ -33,8 +33,8 @@ before conversion or execution. Empty payloads remain inline, and disabling the
 the existing transport behavior.
 
 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.
+for the public payload types. A custom/in-memory store or the optional Azure Blob provider can
+implement storage independently of the transport. Complete the six-phase feature stack before release.
 
 ### Payload cleanup client protocol
 
@@ -44,8 +44,11 @@ the reserved Running orchestration, then signals its batch size. Disabling only
 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.
+the DTS `LargePayloadPurge` service and a compatible worker. Payload-enabled JavaScript workers
+now register the reserved cleanup orchestrator and activities automatically, preserve their
+wildcard filters and bypass business version policy without changing customer filters.
+Deletion verifies ownership and ETags, batches 50 tokens per activity and limits concurrency;
+the runner backs off empty/retry-only batches and preserves configuration through continue-as-new.
 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

packages/durabletask-js-azuremanaged/test/unit/payload-builders.spec.ts

@@ -27,7 +27,7 @@ describe("Azure-managed payload configuration", () => {
     expect(client["_payloads"]).toBeDefined();
     await client.stop();
   });
-  it("preserves worker payload settings and business registrations", () => {
+  it("preserves worker payload settings, business registrations, and internal wildcard filters", () => {
     const worker = new DurableTaskAzureManagedWorkerBuilder()
       .useExternalizedPayloads({ store: new Store() })
       .addNamedOrchestrator("business", () => undefined)
@@ -41,6 +41,6 @@ describe("Azure-managed payload configuration", () => {
         .getWorkitemfilters()!
         .getOrchestrationsList()
         .map((f) => f.getName()),
-    ).toEqual(["business"]);
+    ).toEqual(["business", "BlobPurgeJobOrchestrator"]);
   });
 });

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

@@ -0,0 +1,123 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+import { OrchestrationContext } from "../task/context/orchestration-context";
+import { Task } from "../task/task";
+import { whenAll, whenAny } from "../task";
+import { RetryPolicy } from "../task/retry/retry-policy";
+import { TaskFailureDetails } from "../task/failure-details";
+import { TaskFailedError } from "../task/exception/task-failed-error";
+import { Logger, ConsoleLogger } from "../types/logger.type";
+
+import {
+  BlobPurgeConstants,
+  BlobPurgeJobRunRequest,
+  BlobPurgeOutcome,
+  LargePayloadTombstone,
+  LargePayloadPurgeResult,
+  LargePayloadPurgeDisposition,
+  validatePurgeBatchSize,
+} from "./purge";
+
+export function isUnsupportedPurgeFailure(failure: TaskFailureDetails): boolean {
+  for (let current: TaskFailureDetails | undefined = failure; current; current = current.innerFailure) {
+    if (current.errorType === "System.NotImplementedException" || current.errorType === "NotImplementedException")
+      return true;
+  }
+  return false;
+}
+
+const retry = new RetryPolicy({
+  maxNumberOfAttempts: 3,
+  firstRetryIntervalInMilliseconds: 15000,
+  backoffCoefficient: 2,
+  maxRetryIntervalInMilliseconds: 60000,
+  handleFailure: (failure) => !isUnsupportedPurgeFailure(failure),
+});
+const taskOptions = { retry, version: "" };
+
+export async function* blobPurgeJobOrchestrator(
+  context: OrchestrationContext,
+  input: BlobPurgeJobRunRequest,
+  logger: Logger = new ConsoleLogger(),
+): AsyncGenerator<Task<any>, void, any> {
+  validatePurgeBatchSize(input.PurgeBatchSize);
+  const replayLogger = context.createReplaySafeLogger(logger);
+  let batchSize = input.PurgeBatchSize;
+  let purgedCount = input.PurgedCount ?? 0;
+  let cycles = 0;
+  let configuration = context.waitForExternalEvent(BlobPurgeConstants.configurationEvent);
+  for (;;) {
+    // CAN cannot preserve a configuration already delivered to this waiter.
+    while (configuration.isComplete) {
+      batchSize = configuration.getResult();
+      validatePurgeBatchSize(batchSize);
+      configuration = context.waitForExternalEvent(BlobPurgeConstants.configurationEvent);
+    }
+    if (cycles >= 5) {
+      context.continueAsNew({ PurgeBatchSize: batchSize, PurgedCount: purgedCount }, true);
+      return;
+    }
+    context.setCustomStatus({ Status: "Running", BatchSize: batchSize, PurgedCount: purgedCount });
+    cycles++;
+    let wait = false;
+    try {
+      const tombstones: LargePayloadTombstone[] = yield context.callActivity(
+        BlobPurgeConstants.fetch,
+        batchSize,
+        taskOptions,
+      );
+      if (!tombstones?.length) {
+        wait = true;
+      } else {
+        const results: LargePayloadPurgeResult[] = [];
+        for (let start = 0; start < tombstones.length; start += 50 * 4) {
+          const chunks: LargePayloadTombstone[][] = [];
+          const tasks: Task<BlobPurgeOutcome[]>[] = [];
+          for (let index = start; index < Math.min(start + 50 * 4, tombstones.length); index += 50) {
+            const chunk = tombstones.slice(index, index + 50);
+            chunks.push(chunk);
+            tasks.push(
+              context.callActivity(
+                BlobPurgeConstants.delete,
+                chunk.map((t) => t.PayloadToken),
+                taskOptions,
+              ),
+            );
+          }
+          const outcomes: BlobPurgeOutcome[][] = yield whenAll(tasks);
+          for (let index = 0; index < chunks.length; index++) {
+            const chunk = chunks[index];
+            if (!outcomes[index] || outcomes[index].length !== chunk.length) {
+              throw new Error("The blob delete activity must return exactly one outcome per token.");
+            }
+            results.push(
+              ...chunk.map((tombstone, position) => ({
+                TombstoneToken: tombstone.TombstoneToken,
+                Disposition: outcomes[index][position].Disposition,
+              })),
+            );
+          }
+        }
+        yield context.callActivity(BlobPurgeConstants.report, results, taskOptions);
+        purgedCount += results.filter((r) => r.Disposition === LargePayloadPurgeDisposition.Deleted).length;
+        wait = results.every((r) => r.Disposition === LargePayloadPurgeDisposition.Retry);
+      }
+    } catch (error) {
+      if (error instanceof TaskFailedError && isUnsupportedPurgeFailure(error.details)) {
+        replayLogger.warn("Large payload auto-purge backend unsupported; waiting for explicit configuration.");
+        context.setCustomStatus({ Status: "BackendUnsupported", BatchSize: batchSize, PurgedCount: purgedCount });
+        yield configuration;
+      } else {
+        replayLogger.error("Large payload auto-purge cycle failed; backing off before another attempt.");
+        wait = true;
+      }
+    }
+    if (wait && !configuration.isComplete) {
+      context.setCustomStatus({ Status: "Waiting", BatchSize: batchSize, PurgedCount: purgedCount });
+      const timer = context.createTimer(new Date(context.currentUtcDateTime.getTime() + 60000));
+      yield whenAny([timer, configuration]);
+      if (!timer.isComplete) timer.cancel();
+    }
+  }
+}

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

@@ -0,0 +1,191 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+import * as grpc from "@grpc/grpc-js";
+import * as lp from "../proto/large_payload_purge_pb";
+import { LargePayloadPurgeService } from "../proto/large_payload_purge_grpc_pb";
+import { Registry } from "../worker/registry";
+import { WorkItemFilters } from "../worker/work-item-filters";
+import { Logger } from "../types/logger.type";
+import { MetadataGenerator } from "../utils/grpc-helper.util";
+import { PayloadStore, PayloadDeleteOutcome, PayloadStorageException } from "./payload-store";
+import { runBounded } from "./run-bounded";
+import { callPayloadPurge } from "./purge-transport";
+import {
+  BlobPurgeConstants,
+  BlobPurgeOutcome,
+  LargePayloadPurgeDisposition,
+  LargePayloadPurgeResult,
+  validatePurgeBatchSize,
+} from "./purge";
+import { blobPurgeJobOrchestrator } from "./purge-orchestrator";
+
+const wireDispositions: Record<LargePayloadPurgeDisposition, lp.LargePayloadPurgeDisposition> = {
+  [LargePayloadPurgeDisposition.Unspecified]:
+    lp.LargePayloadPurgeDisposition.LARGE_PAYLOAD_PURGE_DISPOSITION_UNSPECIFIED,
+  [LargePayloadPurgeDisposition.Deleted]: lp.LargePayloadPurgeDisposition.LARGE_PAYLOAD_PURGE_DISPOSITION_DELETED,
+  [LargePayloadPurgeDisposition.Retry]: lp.LargePayloadPurgeDisposition.LARGE_PAYLOAD_PURGE_DISPOSITION_RETRY,
+  [LargePayloadPurgeDisposition.Quarantined]:
+    lp.LargePayloadPurgeDisposition.LARGE_PAYLOAD_PURGE_DISPOSITION_QUARANTINED,
+};
+
+export function registerPayloadPurge(
+  registry: Registry,
+  store: PayloadStore,
+  getClient: () => grpc.Client,
+  metadataGenerator: MetadataGenerator | undefined,
+  getSignal: () => AbortSignal | undefined,
+  logger: Logger,
+): void {
+  registry.addNamedOrchestrator(BlobPurgeConstants.orchestrator, (ctx, input) =>
+    blobPurgeJobOrchestrator(ctx, input, logger),
+  );
+  registry.addNamedActivity(BlobPurgeConstants.fetch, async (_ctx, limit: number) => {
+    validatePurgeBatchSize(limit);
+    try {
+      const response = await callPayloadPurge(
+        getClient(),
+        LargePayloadPurgeService.getLargePayloadTombstones,
+        new lp.GetLargePayloadTombstonesRequest().setLimit(limit),
+        metadataGenerator,
+        getSignal(),
+        new Date(Date.now() + 60000),
+      );
+      return response.getTombstonesList().map((t) => ({
+        TombstoneToken: t.getTombstoneToken(),
+        PayloadToken: t.getPayloadToken(),
+      }));
+    } catch (error) {
+      if (grpcStatus(error) === grpc.status.FAILED_PRECONDITION) {
+        logger.info("Large payload purge fetch precondition not met; idling.");
+        return [];
+      }
+      if (grpcStatus(error) === grpc.status.UNIMPLEMENTED) throw unsupportedBackend(error);
+      throw error;
+    }
+  });
+  registry.addNamedActivity(BlobPurgeConstants.delete, async (_ctx, tokens: string[]) =>
+    deletePayloadChunk(store, tokens, logger, getSignal()),
+  );
+  registry.addNamedActivity(BlobPurgeConstants.report, async (_ctx, results: LargePayloadPurgeResult[]) => {
+    if (!results?.length) return null;
+    const request = new lp.ReportLargePayloadPurgeResultsRequest().setResultsList(
+      results.map((r) => {
+        const disposition = wireDispositions[r.Disposition];
+        if (disposition === undefined) throw new TypeError("Unknown large-payload purge disposition.");
+        return new lp.LargePayloadPurgeResult().setTombstoneToken(r.TombstoneToken).setDisposition(disposition);
+      }),
+    );
+    try {
+      await callPayloadPurge(
+        getClient(),
+        LargePayloadPurgeService.reportLargePayloadPurgeResults,
+        request,
+        metadataGenerator,
+        getSignal(),
+        new Date(Date.now() + 60000),
+      );
+    } catch (error) {
+      if (grpcStatus(error) === grpc.status.UNIMPLEMENTED) throw unsupportedBackend(error);
+      throw error;
+    }
+    return null;
+  });
+}
+
+export async function deletePayloadChunk(
+  store: PayloadStore,
+  tokens: string[],
+  logger: Logger,
+  signal?: AbortSignal,
+): Promise<BlobPurgeOutcome[]> {
+  const outcomes: BlobPurgeOutcome[] = new Array(tokens.length);
+  await runBounded(
+    tokens.map((token, index) => async () => {
+      outcomes[index] = { Disposition: await deletePayload(store, token, logger, signal) };
+    }),
+  );
+  return outcomes;
+}
+
+async function deletePayload(
+  store: PayloadStore,
+  token: string,
+  logger: Logger,
+  signal?: AbortSignal,
+): Promise<LargePayloadPurgeDisposition> {
+  const classify = (disposition: LargePayloadPurgeDisposition, reason: string) => {
+    logger.warn(`Large payload purge: ${reason}; disposition=${disposition}.`);
+    return disposition;
+  };
+  if (token.startsWith("blob:v1:")) return classify(LargePayloadPurgeDisposition.Quarantined, "LegacyV1Token");
+  if (!token.startsWith("blob:v2:")) return classify(LargePayloadPurgeDisposition.Retry, "UnsupportedTokenVersion");
+  const timeout = new AbortController();
+  const timer = setTimeout(() => timeout.abort(), 60000);
+  timer.unref();
+  try {
+    const deleteSignal = signal ? AbortSignal.any([signal, timeout.signal]) : timeout.signal;
+    const outcome = await store.delete(token, deleteSignal);
+    switch (outcome) {
+      case PayloadDeleteOutcome.Deleted:
+      case PayloadDeleteOutcome.AlreadyAbsent:
+        return LargePayloadPurgeDisposition.Deleted;
+      case PayloadDeleteOutcome.NotStoreOwned:
+        logger.info("Large payload purge: NotStoreOwned; blob left untouched.");
+        return LargePayloadPurgeDisposition.Deleted;
+      default:
+        return classify(LargePayloadPurgeDisposition.Retry, "UnexpectedDeleteOutcome");
+    }
+  } catch (error) {
+    if (error instanceof TypeError) return classify(LargePayloadPurgeDisposition.Quarantined, "MalformedToken");
+    if (error instanceof PayloadStorageException)
+      return classify(LargePayloadPurgeDisposition.Retry, "StorageAccountUnreachable");
+    const status = error !== null && typeof error === "object" && "statusCode" in error ? error.statusCode : undefined;
+    if (status === 400) return classify(LargePayloadPurgeDisposition.Quarantined, "InvalidStorageRequest");
+    if (status === 401 || status === 403)
+      return classify(LargePayloadPurgeDisposition.Retry, "StorageAuthorizationFailed");
+    return classify(LargePayloadPurgeDisposition.Retry, "UnconfirmedDelete");
+  } finally {
+    clearTimeout(timer);
+  }
+}
+
+export function isPayloadPurgeTask(name?: string): boolean {
+  return (
+    name === BlobPurgeConstants.orchestrator ||
+    name === BlobPurgeConstants.fetch ||
+    name === BlobPurgeConstants.delete ||
+    name === BlobPurgeConstants.report
+  );
+}
+
+export function withPayloadPurgeFilters(filters: WorkItemFilters): WorkItemFilters {
+  if (!(filters.orchestrations?.length || filters.activities?.length || filters.entities?.length)) return filters;
+  return {
+    ...filters,
+    orchestrations: [
+      ...(filters.orchestrations ?? []).filter((f) => f.name !== BlobPurgeConstants.orchestrator),
+      { name: BlobPurgeConstants.orchestrator, versions: [] },
+    ],
+    activities: [
+      ...(filters.activities ?? []).filter((f) => !isPayloadPurgeTask(f.name)),
+      ...[BlobPurgeConstants.fetch, BlobPurgeConstants.delete, BlobPurgeConstants.report].map((name) => ({
+        name,
+        versions: [],
+      })),
+    ],
+  };
+}
+
+function grpcStatus(error: unknown): unknown {
+  return error !== null && typeof error === "object" && "code" in error ? error.code : undefined;
+}
+
+function unsupportedBackend(cause: unknown): Error {
+  const error = new Error(
+    "The backend does not implement large-payload auto-purge. Upgrade the backend or DTS emulator, then enable auto-purge again.",
+    { cause },
+  );
+  error.name = "System.NotImplementedException";
+  return error;
+}