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.
@@ -50,6 +53,15 @@
 
 ### Fixes
 
+- Scope reserved payload-cleanup version exemptions and filters to their task kind, preserving
+  customer tasks with reserved names in another category and their configured versions.
+- Validate malformed v2 cleanup tokens before store I/O; use `InvalidPayloadTokenError` for
+  provider-specific token failures, including path-style missing blob names. Retry uncertain
+  store failures such as native fetch `TypeError` network errors, not every `TypeError` as malformed.
+- Stop queued cleanup deletes on caller/worker cancellation, drain started deletes and preserve
+  the original cancellation reason. The independent 60-second per-delete timeout remains retryable.
+- Preserve cleanup activity metadata failures unchanged; only scheduler RPC failures trigger
+  precondition idling or unsupported-backend handling.
 - Cancel metadata acquisition waits before dispatch and preserve the original cancellation reason,
   including auto-purge setting and runner lifecycle calls. Cancellation detaches the SDK wait;
   third-party metadata generation may finish later, but cannot dispatch a canceled RPC.

README.md

@@ -45,8 +45,8 @@ propagate unchanged. Completion waits retry scheduler deadlines with fresh metad
 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.
+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
 
@@ -56,8 +56,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-azureblob/CHANGELOG.md

@@ -2,6 +2,9 @@
 
 ### Fixes
 
+- Classify malformed percent/UTF-8 encodings in v2 token paths as `InvalidPayloadTokenError`
+  so cleanup quarantines them without storage I/O. Keep access-scope denials and unrelated
+  storage failures retryable.
 - Default payload reads and deletion to the configured storage endpoint and container, including
   normalized blob paths and path-style account names. Out-of-location deletion references
   remain available for retry. Changing storage locations requires explicit migration or a

packages/durabletask-js-azureblob/src/blob-payload-store.ts

@@ -10,7 +10,12 @@ import {
   RestError,
   StoragePipelineOptions,
 } from "@azure/storage-blob";
-import { PayloadStore, PayloadStorageException, PayloadDeleteOutcome } from "@microsoft/durabletask-js";
+import {
+  InvalidPayloadTokenError,
+  PayloadStore,
+  PayloadStorageException,
+  PayloadDeleteOutcome,
+} from "@microsoft/durabletask-js";
 import { randomUUID } from "crypto";
 import { Writable } from "stream";
 import { pipeline } from "stream/promises";
@@ -255,7 +260,7 @@ export class BlobPayloadStore extends PayloadStore {
       const rest = token.slice("blob:v1:".length);
       const separator = rest.indexOf(":");
       if (separator < 1 || separator === rest.length - 1) {
-        throw new TypeError("Invalid external payload token.");
+        throw new InvalidPayloadTokenError("Invalid external payload token.");
       }
       const containerName = rest.slice(0, separator);
       const primary = this._primaryLocation;
@@ -266,7 +271,7 @@ export class BlobPayloadStore extends PayloadStore {
       this._validateBlobUrl(new URL(blob.url), primary, containerName);
       return blob;
     }
-    if (!token.startsWith("blob:v2:")) throw new TypeError("Invalid external payload token.");
+    if (!token.startsWith("blob:v2:")) throw new InvalidPayloadTokenError("Invalid external payload token.");
     const value = token.slice("blob:v2:".length).split(/[?#]/, 1)[0];
     if (/\/(?:\.|%2e){1,2}(?=\/|$)/i.test(value)) {
       throw new PayloadStorageException("The payload blob URL must not contain dot-segment traversal.");
@@ -275,21 +280,22 @@ export class BlobPayloadStore extends PayloadStore {
     try {
       url = new URL(token.slice("blob:v2:".length));
     } catch (error) {
-      throw new TypeError("Invalid external payload blob URL.", { cause: error });
+      throw new InvalidPayloadTokenError("Invalid external payload blob URL.", { cause: error });
     }
     if ((url.protocol !== "https:" && url.protocol !== "http:") || url.username || url.password) {
-      throw new TypeError("Invalid external payload blob URL.");
+      throw new InvalidPayloadTokenError("Invalid external payload blob URL.");
     }
     url.search = "";
     url.hash = "";
     const location = this._locations.find((candidate) => this._relativeBlobPath(url, candidate) !== undefined);
     if (!location) throw new PayloadStorageException("The payload blob URL is outside the configured access scope.");
     const relative = this._relativeBlobPath(url, location)!;
     const separator = relative.indexOf("/");
-    if (separator < 1 || separator === relative.length - 1) throw new TypeError("Invalid external payload blob URL.");
+    if (separator < 1 || separator === relative.length - 1)
+      throw new InvalidPayloadTokenError("Invalid external payload blob URL.");
     this._validateBlobUrl(url, location);
-    const containerName = decodeURIComponent(relative.slice(0, separator));
-    const name = decodeURIComponent(relative.slice(separator + 1));
+    const containerName = decodeTokenComponent(relative.slice(0, separator));
+    const name = decodeTokenComponent(relative.slice(separator + 1));
     if (name.split("/").some((segment) => segment === "." || segment === "..")) {
       throw new PayloadStorageException("The payload blob name must not contain dot segments.");
     }
@@ -312,7 +318,7 @@ export class BlobPayloadStore extends PayloadStore {
     const separator = relative?.indexOf("/") ?? -1;
     const container =
       relative !== undefined && separator > 0
-        ? encodeURIComponent(decodeURIComponent(relative.slice(0, separator)))
+        ? encodeURIComponent(decodeTokenComponent(relative.slice(0, separator)))
         : undefined;
     if (
       relative === undefined ||
@@ -355,6 +361,17 @@ export class BlobPayloadStore extends PayloadStore {
   }
 }
 
+function decodeTokenComponent(value: string): string {
+  try {
+    return decodeURIComponent(value);
+  } catch (error) {
+    if (error instanceof URIError) {
+      throw new InvalidPayloadTokenError("Invalid external payload blob URL encoding.", { cause: error });
+    }
+    throw error;
+  }
+}
+
 function trimTrailingSlashes(value: string): string {
   let end = value.length;
   while (end > 0 && value.charCodeAt(end - 1) === 47) end--;

packages/durabletask-js-azureblob/test/blob-payload-store.spec.ts

@@ -13,7 +13,13 @@ import {
   StorageRetryOptions,
   StorageRetryPolicyType,
 } from "@azure/storage-blob";
-import { PayloadDeleteOutcome, PayloadStorageException } from "@microsoft/durabletask-js";
+import {
+  InvalidPayloadTokenError,
+  NoOpLogger,
+  PayloadDeleteOutcome,
+  PayloadStorageException,
+} from "@microsoft/durabletask-js";
+import { deletePayloadChunk } from "../../durabletask-js/src/payloads/purge-worker";
 import type { StringValue } from "google-protobuf/google/protobuf/wrappers_pb";
 import { PayloadProcessor } from "../../durabletask-js/src/payloads/payload-processor";
 import * as pb from "../../durabletask-js/src/proto/orchestrator_service_pb";
@@ -88,6 +94,109 @@ describe("BlobPayloadStore", () => {
     http = new RecordingClient();
   });
 
+  it.each(["127.0.0.1", "localhost"])(
+    "quarantines the real provider's missing path-style blob name without storage I/O (%s)",
+    async (host) => {
+      const endpoint = `http://${host}:10000/devstoreaccount1`;
+      const store = new BlobPayloadStore({
+        connectionString: `DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;AccountKey=${accountKey};BlobEndpoint=${endpoint}`,
+        clientOptions: { httpClient: http },
+      });
+      const token = `blob:v2:${endpoint}/payloads/`;
+      await expect(store.delete(token)).rejects.toBeInstanceOf(InvalidPayloadTokenError);
+      await expect(deletePayloadChunk(store, [token], new NoOpLogger())).resolves.toEqual([{ Disposition: 3 }]);
+      expect(http.requests).toHaveLength(0);
+    },
+  );
+
+  it.each(["blob:v2:https://account.invalid/archive/blob", "blob:v2:https://other.invalid/archive/blob"])(
+    "keeps valid out-of-scope cleanup references retryable until explicitly authorized (%#)",
+    async (token) => {
+      const clientOptions = { httpClient: http, retryOptions: { maxTries: 1 } };
+      const restricted = new BlobPayloadStore({ connectionString, clientOptions });
+      const failure: unknown = await restricted.delete(token).catch((error: unknown) => error);
+      expect(failure).toBeInstanceOf(PayloadStorageException);
+      expect(failure).not.toBeInstanceOf(InvalidPayloadTokenError);
+      await expect(deletePayloadChunk(restricted, [token], new NoOpLogger())).resolves.toEqual([{ Disposition: 2 }]);
+      expect(http.requests).toHaveLength(0);
+
+      const authorized = token.includes("other.invalid")
+        ? new BlobPayloadStore({
+            accountUri: "https://account.invalid",
+            credential: {
+              getToken: async () => ({ token: "synthetic-token", expiresOnTimestamp: Date.now() + 3600000 }),
+            },
+            additionalAccountUris: ["https://other.invalid"],
+            clientOptions,
+          })
+        : new BlobPayloadStore({ connectionString, allowAllContainersInConfiguredAccount: true, clientOptions });
+      await expect(deletePayloadChunk(authorized, [token], new NoOpLogger())).resolves.toEqual([{ Disposition: 1 }]);
+      expect(http.requests.map((request) => request.method)).toEqual(["HEAD", "DELETE"]);
+      expect(http.requests[1].headers.get("if-match")).toBe('"owned-etag"');
+    },
+  );
+
+  describe.each([
+    { endpoint: "https://account.invalid", connectionString },
+    {
+      endpoint: "http://127.0.0.1:10000/devstoreaccount1",
+      connectionString: `DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;AccountKey=${accountKey};BlobEndpoint=http://127.0.0.1:10000/devstoreaccount1`,
+    },
+  ])("token URI decoding ($endpoint)", ({ endpoint, connectionString }) => {
+    const malformedPaths = [
+      "durabletask-payloads/%FF",
+      "durabletask-payloads/%",
+      "durabletask-payloads/%E2%82",
+      "%FF/blob",
+      "%/blob",
+      "%E2%82/blob",
+    ];
+    const make = () =>
+      new BlobPayloadStore({
+        connectionString,
+        clientOptions: { httpClient: http, retryOptions: { maxTries: 1 } },
+      });
+
+    it.each(malformedPaths)("reports malformed encoding as a typed token failure (%s)", async (path) => {
+      const failure: unknown = await make()
+        .delete(`blob:v2:${endpoint}/${path}`)
+        .catch((error: unknown) => error);
+      expect(http.requests).toHaveLength(0);
+      expect(failure).toBeInstanceOf(InvalidPayloadTokenError);
+      expect(failure).toHaveProperty("cause", expect.any(URIError));
+    });
+
+    it.each(malformedPaths)("quarantines malformed encoding without storage I/O (%s)", async (path) => {
+      const result = await deletePayloadChunk(make(), [`blob:v2:${endpoint}/${path}`], new NoOpLogger());
+      expect(http.requests).toHaveLength(0);
+      expect(result).toEqual([{ Disposition: 3 }]);
+    });
+
+    it.each([
+      ["/durabletask%2Dpayloads/folder%2Fnested%2F%25FF%2F100%25", "/durabletask-payloads/folder/nested/%25FF/100%25"],
+      ["/durabletask-payloads/folder/a%252Fb%20%25", "/durabletask-payloads/folder/a%252Fb%20%25"],
+    ])("preserves valid percent and nested names through conditional deletion (%s)", async (path, expected) => {
+      const store = make();
+      const token = `blob:v2:${endpoint}${path}`;
+      await expect(store.download(token)).resolves.toBe("payload");
+      await expect(deletePayloadChunk(store, [token], new NoOpLogger())).resolves.toEqual([{ Disposition: 1 }]);
+      expect(http.requests.map((request) => request.method)).toEqual(["GET", "HEAD", "DELETE"]);
+      for (const request of http.requests) {
+        const url = new URL(request.url);
+        expect(url.origin + url.pathname).toBe(`${endpoint}${expected}`);
+      }
+      expect(http.requests[2].headers.get("if-match")).toBe('"owned-etag"');
+    });
+
+    it("keeps a transport URIError retryable rather than quarantining it", async () => {
+      http.failures.push(new URIError("Synthetic transport failure"));
+      await expect(
+        deletePayloadChunk(make(), [`blob:v2:${endpoint}/durabletask-payloads/blob`], new NoOpLogger()),
+      ).resolves.toEqual([{ Disposition: 2 }]);
+      expect(http.requests.map((request) => request.method)).toEqual(["HEAD"]);
+    });
+  });
+
   describe("BlobPayloadStore native SDK retries and options", () => {
     let http: RecordingClient;
     const token = "blob:v2:https://account.invalid/durabletask-payloads/blob";

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/index.ts

@@ -4,6 +4,7 @@
 // Client and Worker
 export {
   PayloadStore,
+  InvalidPayloadTokenError,
   PayloadStorageOptions,
   PayloadStorageException,
   PayloadDeleteOutcome,

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

@@ -15,12 +15,23 @@ export abstract class PayloadStore {
   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. */
+  /**
+   * Optional cleanup support. Never delete a blob without verifying ownership and its ETag.
+   * Use InvalidPayloadTokenError only for deterministic token-validation failures.
+   */
   async delete(_token: string, _signal?: AbortSignal): Promise<PayloadDeleteOutcome> {
     throw new Error("This payload store does not support deletion.");
   }
 }
 
+/** A deterministic malformed storage token, distinct from uncertain I/O TypeErrors. */
+export class InvalidPayloadTokenError extends TypeError {
+  constructor(message: string, options?: { cause?: unknown }) {
+    super(message, options);
+    this.name = "InvalidPayloadTokenError";
+  }
+}
+
 /** A permanent payload storage or size-limit failure. */
 export class PayloadStorageException extends Error {
   constructor(message: string, options?: { cause?: unknown }) {