diff --git a/CHANGELOG.md b/CHANGELOG.md index 3f2f3814..10796fa2 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/README.md b/README.md index 7c4c21ed..5fcdcdaa 100644 --- a/README.md +++ b/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 diff --git a/packages/durabletask-js-azureblob/CHANGELOG.md b/packages/durabletask-js-azureblob/CHANGELOG.md index 7307a80e..539b28c3 100644 --- a/packages/durabletask-js-azureblob/CHANGELOG.md +++ b/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 diff --git a/packages/durabletask-js-azureblob/src/blob-payload-store.ts b/packages/durabletask-js-azureblob/src/blob-payload-store.ts index 99465271..71214be7 100644 --- a/packages/durabletask-js-azureblob/src/blob-payload-store.ts +++ b/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,10 +280,10 @@ 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 = ""; @@ -286,10 +291,11 @@ export class BlobPayloadStore extends PayloadStore { 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--; diff --git a/packages/durabletask-js-azureblob/test/blob-payload-store.spec.ts b/packages/durabletask-js-azureblob/test/blob-payload-store.spec.ts index b960f2a5..258accc3 100644 --- a/packages/durabletask-js-azureblob/test/blob-payload-store.spec.ts +++ b/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"; diff --git a/packages/durabletask-js-azuremanaged/test/unit/payload-builders.spec.ts b/packages/durabletask-js-azuremanaged/test/unit/payload-builders.spec.ts index d7975126..a9b191c0 100644 --- a/packages/durabletask-js-azuremanaged/test/unit/payload-builders.spec.ts +++ b/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"]); }); }); diff --git a/packages/durabletask-js/src/index.ts b/packages/durabletask-js/src/index.ts index f319c996..206a7757 100644 --- a/packages/durabletask-js/src/index.ts +++ b/packages/durabletask-js/src/index.ts @@ -4,6 +4,7 @@ // Client and Worker export { PayloadStore, + InvalidPayloadTokenError, PayloadStorageOptions, PayloadStorageException, PayloadDeleteOutcome, diff --git a/packages/durabletask-js/src/payloads/payload-store.ts b/packages/durabletask-js/src/payloads/payload-store.ts index 2eafae82..be15d281 100644 --- a/packages/durabletask-js/src/payloads/payload-store.ts +++ b/packages/durabletask-js/src/payloads/payload-store.ts @@ -15,12 +15,23 @@ export abstract class PayloadStore { abstract download(token: string, signal?: AbortSignal): Promise; 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 { 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 }) { diff --git a/packages/durabletask-js/src/payloads/purge-orchestrator.ts b/packages/durabletask-js/src/payloads/purge-orchestrator.ts new file mode 100644 index 00000000..0484c626 --- /dev/null +++ b/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, 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[] = []; + 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(); + } + } +} diff --git a/packages/durabletask-js/src/payloads/purge-worker.ts b/packages/durabletask-js/src/payloads/purge-worker.ts new file mode 100644 index 00000000..ce1f41aa --- /dev/null +++ b/packages/durabletask-js/src/payloads/purge-worker.ts @@ -0,0 +1,218 @@ +// 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 { getMetadata, MetadataGenerator } from "../utils/grpc-helper.util"; +import { InvalidPayloadTokenError, 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.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); + const signal = getSignal(); + const metadata = metadataGenerator ? await getMetadata(metadataGenerator, signal) : undefined; + const client = getClient(); + let response: lp.GetLargePayloadTombstonesResponse; + try { + response = await callPayloadPurge( + client, + LargePayloadPurgeService.getLargePayloadTombstones, + new lp.GetLargePayloadTombstonesRequest().setLimit(limit), + metadata ? async () => metadata : undefined, + signal, + new Date(Date.now() + 60000), + ); + } catch (error) { + signal?.throwIfAborted(); + 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; + } + return response.getTombstonesList().map((t) => ({ + TombstoneToken: t.getTombstoneToken(), + PayloadToken: t.getPayloadToken(), + })); + }); + 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); + }), + ); + const signal = getSignal(); + const metadata = metadataGenerator ? await getMetadata(metadataGenerator, signal) : undefined; + const client = getClient(); + try { + await callPayloadPurge( + client, + LargePayloadPurgeService.reportLargePayloadPurgeResults, + request, + metadata ? async () => metadata : undefined, + signal, + new Date(Date.now() + 60000), + ); + } catch (error) { + signal?.throwIfAborted(); + 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 { + signal?.throwIfAborted(); + const outcomes: BlobPurgeOutcome[] = new Array(tokens.length); + await runBounded( + tokens.map((token, index) => async () => { + outcomes[index] = { Disposition: await deletePayload(store, token, logger, signal) }; + }), + signal, + ); + signal?.throwIfAborted(); + return outcomes; +} + +async function deletePayload( + store: PayloadStore, + token: string, + logger: Logger, + signal?: AbortSignal, +): Promise { + signal?.throwIfAborted(); + 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"); + let url: URL; + try { + url = new URL(token.slice("blob:v2:".length)); + } catch { + return classify(LargePayloadPurgeDisposition.Quarantined, "MalformedToken"); + } + if ( + (url.protocol !== "http:" && url.protocol !== "https:") || + url.username || + url.password || + !/^\/[^/]+\/.+/.test(url.pathname) + ) { + return classify(LargePayloadPurgeDisposition.Quarantined, "MalformedToken"); + } + 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); + signal?.throwIfAborted(); + 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) { + signal?.throwIfAborted(); + if (error instanceof InvalidPayloadTokenError) + 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 | undefined, kind: "orchestrator" | "activity"): boolean { + return kind === "orchestrator" + ? 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) => !isPayloadPurgeTask(f.name, "orchestrator")), + { name: BlobPurgeConstants.orchestrator, versions: [] }, + ], + activities: [ + ...(filters.activities ?? []).filter((f) => !isPayloadPurgeTask(f.name, "activity")), + ...[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; +} diff --git a/packages/durabletask-js/src/worker/task-hub-grpc-worker.ts b/packages/durabletask-js/src/worker/task-hub-grpc-worker.ts index b5b8d35b..84886083 100644 --- a/packages/durabletask-js/src/worker/task-hub-grpc-worker.ts +++ b/packages/durabletask-js/src/worker/task-hub-grpc-worker.ts @@ -29,6 +29,7 @@ import { compareVersions } from "../utils/versioning.util"; import { PayloadStorageOptions } from "../payloads/payload-store"; import { PayloadProcessor } from "../payloads/payload-processor"; import { readPayloadHistory, PayloadHydrationError } from "../payloads/history-stream"; +import { registerPayloadPurge, withPayloadPurgeFilters, isPayloadPurgeTask } from "../payloads/purge-worker"; import * as WorkerLogs from "./logs"; import { ConcurrencyOptions, ResolvedConcurrencyOptions, resolveConcurrencyOptions } from "./concurrency-options"; import { @@ -266,6 +267,19 @@ export class TaskHubGrpcWorker { this._abortController = null; this._workerLoopPromise = null; this._deferredStubCloseTimers = new Map(); + if (typeof hostAddressOrOptions === "object" && hostAddressOrOptions?.payloads) { + registerPayloadPurge( + this._registry, + hostAddressOrOptions.payloads.store, + () => { + if (!this._stub) throw new Error("The payload purge transport is not connected."); + return this._stub; + }, + this._metadataGenerator, + () => this._abortController?.signal, + this._logger, + ); + } } /** The configured maximum backend timer interval in milliseconds. */ @@ -880,7 +894,7 @@ export class TaskHubGrpcWorker { this._workItemFilters === "auto" ? generateWorkItemFiltersFromRegistry(this._registry, this._versioning) : this._workItemFilters; - request.setWorkitemfilters(toGrpcWorkItemFilters(filters)); + request.setWorkitemfilters(toGrpcWorkItemFilters(this._payloads ? withPayloadPurgeFilters(filters) : filters)); } return request; @@ -889,13 +903,18 @@ export class TaskHubGrpcWorker { /** * Result of version compatibility check. */ - private _checkVersionCompatibility(orchestrationVersion?: string): { + private _checkVersionCompatibility( + orchestrationVersion: string | undefined, + taskName: string | undefined, + taskKind: "orchestrator" | "activity", + ): { compatible: boolean; shouldFail: boolean; orchestrationVersion?: string; errorType?: string; errorMessage?: string; } { + if (this._payloads && isPayloadPurgeTask(taskName, taskKind)) return { compatible: true, shouldFail: false }; // If no versioning options configured or match strategy is None, always compatible if ( !this._versioning || @@ -1158,7 +1177,15 @@ export class TaskHubGrpcWorker { } // Check version compatibility if versioning is enabled - const versionCheckResult = this._checkVersionCompatibility(this._getOrchestrationVersion(req)); + const taskName = [...req.getPasteventsList(), ...req.getNeweventsList()] + .find((event) => event.hasExecutionstarted()) + ?.getExecutionstarted() + ?.getName(); + const versionCheckResult = this._checkVersionCompatibility( + this._getOrchestrationVersion(req), + taskName, + "orchestrator", + ); if (!versionCheckResult.compatible) { if (versionCheckResult.shouldFail) { // Fail the orchestration with version mismatch error @@ -1382,7 +1409,7 @@ export class TaskHubGrpcWorker { throw new Error("Activity request does not contain an orchestration instance id"); } - const versionCheck = this._checkVersionCompatibility(req.getVersion()?.getValue()); + const versionCheck = this._checkVersionCompatibility(req.getVersion()?.getValue(), req.getName(), "activity"); if (!versionCheck.compatible && !versionCheck.shouldFail) { WorkerLogs.versionMismatchAbandon(this._logger, instanceId, versionCheck.errorType!, versionCheck.errorMessage!); try { diff --git a/packages/durabletask-js/test/payload-purge-worker-transport.spec.ts b/packages/durabletask-js/test/payload-purge-worker-transport.spec.ts new file mode 100644 index 00000000..a20607d8 --- /dev/null +++ b/packages/durabletask-js/test/payload-purge-worker-transport.spec.ts @@ -0,0 +1,202 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import * as grpc from "@grpc/grpc-js"; +import { TaskHubGrpcWorker, NoOpLogger, PayloadStore, ActivityContext } from "../src"; +import * as stubs from "../src/proto/orchestrator_service_grpc_pb"; +import * as lp from "../src/proto/large_payload_purge_pb"; +import { LargePayloadPurgeService, ILargePayloadPurgeServer } from "../src/proto/large_payload_purge_grpc_pb"; +import { BlobPurgeConstants } from "../src/payloads/purge"; +import { Registry } from "../src/worker/registry"; +import { registerPayloadPurge } from "../src/payloads/purge-worker"; + +class Store extends PayloadStore { + async upload(): Promise { + return "blob:v2:https://account.invalid/c/blob"; + } + async download(): Promise { + return "null"; + } + isKnownPayloadToken(value: string): boolean { + return value.startsWith("blob:"); + } +} + +it("uses the current authenticated sidecar transport for reserved fetch/report activities", async () => { + const calls: string[] = []; + let rpcError: grpc.status | undefined; + const server = new grpc.Server(); + const service = { + getLargePayloadTombstones: (call, callback) => { + calls.push("fetch"); + expect(call.metadata.get("taskhub")).toEqual(["payload-test"]); + expect(Number(call.getDeadline()) - Date.now()).toBeLessThanOrEqual(61000); + if (rpcError !== undefined) callback({ code: rpcError, message: "backend" }); + else + callback( + null, + new lp.GetLargePayloadTombstonesResponse().setTombstonesList([ + new lp.LargePayloadTombstone() + .setTombstoneToken("opaque:\u00e9:do-not-parse") + .setPayloadToken("blob:v2:https://a.invalid/c/b"), + ]), + ); + }, + reportLargePayloadPurgeResults: (call, callback) => { + calls.push("report"); + expect(call.metadata.get("taskhub")).toEqual(["payload-test"]); + expect(Number(call.getDeadline()) - Date.now()).toBeLessThanOrEqual(61000); + expect(call.request.getResultsList()[0].getTombstoneToken()).toBe("opaque:\u00e9:do-not-parse"); + callback(null, new lp.ReportLargePayloadPurgeResultsResponse()); + }, + } satisfies Pick; + server.addService(LargePayloadPurgeService, service); + const port = await new Promise((resolve, reject) => + server.bindAsync("127.0.0.1:0", grpc.ServerCredentials.createInsecure(), (error, port) => + error ? reject(error) : resolve(port), + ), + ); + const first = new stubs.TaskHubSidecarServiceClient(`127.0.0.1:${port}`, grpc.credentials.createInsecure()); + const replacement = new stubs.TaskHubSidecarServiceClient(`127.0.0.1:${port}`, grpc.credentials.createInsecure()); + const firstCall = jest.spyOn(first, "makeUnaryRequest"); + const replacementCall = jest.spyOn(replacement, "makeUnaryRequest"); + try { + const worker = new TaskHubGrpcWorker({ + payloads: { store: new Store() }, + logger: new NoOpLogger(), + metadataGenerator: async () => { + const metadata = new grpc.Metadata(); + metadata.set("taskhub", "payload-test"); + return metadata; + }, + }); + + worker["_stub"] = first; + const fetch = worker["_registry"].getActivity(BlobPurgeConstants.fetch)!; + const report = worker["_registry"].getActivity(BlobPurgeConstants.report)!; + const tombstones = await fetch(new ActivityContext("job", 0), 10); + expect(tombstones).toEqual([ + { TombstoneToken: "opaque:\u00e9:do-not-parse", PayloadToken: "blob:v2:https://a.invalid/c/b" }, + ]); + worker["_stub"] = replacement; + await report(new ActivityContext("job", 1), [{ TombstoneToken: tombstones[0].TombstoneToken, Disposition: 1 }]); + expect(firstCall).toHaveBeenCalledTimes(1); + expect(replacementCall).toHaveBeenCalledTimes(1); + expect(calls).toEqual(["fetch", "report"]); + rpcError = grpc.status.FAILED_PRECONDITION; + expect(await fetch(new ActivityContext("job", 2), 10)).toEqual([]); + rpcError = grpc.status.UNIMPLEMENTED; + await expect(fetch(new ActivityContext("job", 3), 10)).rejects.toMatchObject({ + name: "System.NotImplementedException", + }); + } finally { + first.close(); + replacement.close(); + server.forceShutdown(); + } +}); + +describe("purge activities metadata and RPC error boundaries", () => { + let server: grpc.Server; + let address: string; + let client: stubs.TaskHubSidecarServiceClient; + let registry: Registry; + let controller: AbortController; + let calls: string[]; + let rpcError: grpc.status | undefined; + let metadata: jest.Mock, []>; + + beforeAll(async () => { + server = new grpc.Server(); + server.addService(LargePayloadPurgeService, { + getLargePayloadTombstones: (_call, callback) => { + calls.push("fetch"); + if (rpcError !== undefined) callback({ code: rpcError, message: "Backend response" }); + else callback(null, new lp.GetLargePayloadTombstonesResponse()); + }, + reportLargePayloadPurgeResults: (_call, callback) => { + calls.push("report"); + if (rpcError !== undefined) callback({ code: rpcError, message: "Backend response" }); + else callback(null, new lp.ReportLargePayloadPurgeResultsResponse()); + }, + } satisfies Pick); + const port = await new Promise((resolve, reject) => + server.bindAsync("127.0.0.1:0", grpc.ServerCredentials.createInsecure(), (error, port) => + error ? reject(error) : resolve(port), + ), + ); + address = `127.0.0.1:${port}`; + }); + + beforeEach(() => { + calls = []; + rpcError = undefined; + controller = new AbortController(); + metadata = jest.fn(async () => new grpc.Metadata()); + client = new stubs.TaskHubSidecarServiceClient(address, grpc.credentials.createInsecure()); + registry = new Registry(); + registerPayloadPurge( + registry, + new Store(), + () => client, + metadata, + () => controller.signal, + new NoOpLogger(), + ); + }); + + afterEach(() => client.close()); + afterAll(() => server.forceShutdown()); + + const invoke = (kind: "fetch" | "report") => { + const fn = registry.getActivity(kind === "fetch" ? BlobPurgeConstants.fetch : BlobPurgeConstants.report)!; + return Promise.resolve( + fn(new ActivityContext("job", 1), kind === "fetch" ? 50 : [{ TombstoneToken: "opaque", Disposition: 2 }]), + ); + }; + + describe.each(["fetch", "report"] as const)("%s", (kind) => { + it.each([grpc.status.FAILED_PRECONDITION, grpc.status.UNIMPLEMENTED, grpc.status.UNAVAILABLE])( + "preserves metadata error code%s by identity without dispatching a scheduler RPC", + async (code) => { + const failure = Object.freeze(Object.assign(new Error("Metadata dependency failed"), { code })); + const rpc = jest.spyOn(client, "makeUnaryRequest"); + metadata.mockRejectedValue(failure); + await expect(invoke(kind)).rejects.toBe(failure); + expect(metadata).toHaveBeenCalledTimes(1); + expect(rpc).not.toHaveBeenCalled(); + expect(calls).toEqual([]); + }, + ); + + it.each([grpc.status.FAILED_PRECONDITION, grpc.status.UNIMPLEMENTED])( + "preserves a caller cancellation shaped like backend status%s", + async (code) => { + const reason = Object.freeze(Object.assign(new Error("Caller canceled"), { code })); + controller.abort(reason); + await expect(invoke(kind)).rejects.toBe(reason); + expect(metadata).not.toHaveBeenCalled(); + expect(calls).toEqual([]); + }, + ); + + it.each([grpc.status.FAILED_PRECONDITION, grpc.status.UNIMPLEMENTED, grpc.status.UNAVAILABLE])( + "classifies only an actual scheduler RPC status%s", + async (code) => { + rpcError = code; + if (kind === "fetch" && code === grpc.status.FAILED_PRECONDITION) { + await expect(invoke(kind)).resolves.toEqual([]); + } else if (code === grpc.status.UNIMPLEMENTED) { + await expect(invoke(kind)).rejects.toMatchObject({ + name: "System.NotImplementedException", + cause: { code }, + }); + } else { + await expect(invoke(kind)).rejects.toMatchObject({ code }); + } + expect(metadata).toHaveBeenCalledTimes(1); + expect(calls).toEqual([kind]); + }, + ); + }); +}); diff --git a/packages/durabletask-js/test/payload-purge-worker.spec.ts b/packages/durabletask-js/test/payload-purge-worker.spec.ts new file mode 100644 index 00000000..ebde6065 --- /dev/null +++ b/packages/durabletask-js/test/payload-purge-worker.spec.ts @@ -0,0 +1,743 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { + TaskHubGrpcWorker, + NoOpLogger, + PayloadStore, + PayloadDeleteOutcome, + VersionMatchStrategy, + VersionFailureStrategy, + WorkItemFilters, + TaskFailedError, +} from "../src"; +import * as grpc from "@grpc/grpc-js"; +import * as pb from "../src/proto/orchestrator_service_pb"; +import * as stubs from "../src/proto/orchestrator_service_grpc_pb"; +import * as lp from "../src/proto/large_payload_purge_pb"; +import { StringValue } from "google-protobuf/google/protobuf/wrappers_pb"; +import { newExecutionStartedEvent, newOrchestratorStartedEvent } from "../src/utils/pb-helper.util"; +import { BlobPurgeConstants, BlobPurgeOutcome, LargePayloadPurgeDisposition } from "../src/payloads/purge"; +import { blobPurgeJobOrchestrator } from "../src/payloads/purge-orchestrator"; +import { deletePayloadChunk } from "../src/payloads/purge-worker"; +import { RuntimeOrchestrationContext } from "../src/worker/runtime-orchestration-context"; +import { RetryPolicy } from "../src/task/retry/retry-policy"; +import { EventEmitter, getEventListeners } from "events"; +import { createServer } from "net"; +import { execFile } from "child_process"; +import { join } from "path"; +import { promisify } from "util"; + +class Store extends PayloadStore { + async upload(): Promise { + return "blob:v2:https://account.invalid/payloads/blob"; + } + async download(): Promise { + return "null"; + } + isKnownPayloadToken(value: string): boolean { + return value.startsWith("blob:"); + } + async delete(_token: string, _signal?: AbortSignal): Promise { + return PayloadDeleteOutcome.Deleted; + } +} +const logger = new NoOpLogger(); + +describe("payload purge worker registrations", () => { + it.each([{}, { orchestrations: [], activities: [], entities: [] }])( + "leaves all-empty filters unfiltered", + (filters) => { + const worker = new TaskHubGrpcWorker({ payloads: { store: new Store() }, workItemFilters: filters }); + expect(worker["_buildGetWorkItemsRequest"]().getWorkitemfilters()!.toObject()).toEqual({ + orchestrationsList: [], + activitiesList: [], + entitiesList: [], + }); + }, + ); + it("augments entity-only filters, bypasses business versions, and does not mutate caller lists", () => { + const filters: WorkItemFilters = { entities: [{ name: "counter" }] }; + const worker = new TaskHubGrpcWorker({ + payloads: { store: new Store() }, + workItemFilters: filters, + versioning: { version: "business", matchStrategy: VersionMatchStrategy.Strict }, + }); + const wire = worker["_buildGetWorkItemsRequest"]().getWorkitemfilters()!; + expect(wire.getOrchestrationsList().map((f) => [f.getName(), f.getVersionsList()])).toEqual([ + [BlobPurgeConstants.orchestrator, []], + ]); + expect(wire.getActivitiesList()).toHaveLength(3); + expect(filters).toEqual({ entities: [{ name: "counter" }] }); + expect(worker["_checkVersionCompatibility"]("", BlobPurgeConstants.orchestrator, "orchestrator").compatible).toBe( + true, + ); + expect(worker["_checkVersionCompatibility"]("", "business", "orchestrator").compatible).toBe(false); + }); + + it.each([ + { automatic: false, payloads: true }, + { automatic: true, payloads: true }, + { automatic: false, payloads: false }, + { automatic: true, payloads: false }, + ])( + "preserves wrong-kind customer filter names/versions (auto=$automatic,payloads=$payloads)", + ({ automatic, payloads }) => { + const activities = [BlobPurgeConstants.fetch, BlobPurgeConstants.delete, BlobPurgeConstants.report]; + const filters: WorkItemFilters = { + orchestrations: [...activities, BlobPurgeConstants.orchestrator].map((name) => ({ + name, + versions: ["customer-version"], + })), + activities: [BlobPurgeConstants.orchestrator, ...activities].map((name) => ({ + name, + versions: ["customer-version"], + })), + }; + const original = JSON.parse(JSON.stringify(filters)); + const worker = new TaskHubGrpcWorker({ + payloads: payloads ? { store: new Store() } : undefined, + workItemFilters: automatic ? "auto" : filters, + versioning: { version: "2.0", matchStrategy: VersionMatchStrategy.Strict }, + }); + for (const { name } of filters.orchestrations!) { + if (!worker["_registry"].getOrchestrator(name)) worker.addNamedOrchestrator(name, async () => {}); + } + for (const { name } of filters.activities!) { + if (!worker["_registry"].getActivity(name)) worker.addNamedActivity(name, () => {}); + } + const wire = worker["_buildGetWorkItemsRequest"]().getWorkitemfilters()!; + const customerVersions = automatic ? ["2.0"] : ["customer-version"]; + expect(wire.getOrchestrationsList().map((filter) => [filter.getName(), filter.getVersionsList()])).toEqual([ + ...activities.map((name) => [name, customerVersions]), + [BlobPurgeConstants.orchestrator, payloads ? [] : customerVersions], + ]); + expect(wire.getActivitiesList().map((filter) => [filter.getName(), filter.getVersionsList()])).toEqual([ + [BlobPurgeConstants.orchestrator, customerVersions], + ...activities.map((name) => [name, payloads ? [] : customerVersions]), + ]); + expect(filters).toEqual(original); + }, + ); + + it.each([undefined, {}])("does not add filters to an unfiltered payload worker (%#)", (filters) => { + const worker = new TaskHubGrpcWorker({ payloads: { store: new Store() }, workItemFilters: filters }); + const wire = worker["_buildGetWorkItemsRequest"]().getWorkitemfilters(); + expect(wire?.getOrchestrationsList() ?? []).toEqual([]); + expect(wire?.getActivitiesList() ?? []).toEqual([]); + }); +}); + +describe("payload purge task-kind version dispatch", () => { + type Kind = "activity" | "orchestrator"; + const reserved: { kind: Kind; name: string }[] = [ + { kind: "orchestrator", name: BlobPurgeConstants.orchestrator }, + ...[BlobPurgeConstants.fetch, BlobPurgeConstants.delete, BlobPurgeConstants.report].map((name) => ({ + kind: "activity" as const, + name, + })), + ]; + const wrongKind: { kind: Kind; name: string }[] = reserved.map(({ kind, name }) => ({ + kind: kind === "activity" ? "orchestrator" : "activity", + name, + })); + const customerTasks: { kind: Kind; name: string }[] = [ + ...wrongKind, + { kind: "activity", name: "business-activity" }, + { kind: "orchestrator", name: "business-orchestration" }, + { kind: "activity", name: `${BlobPurgeConstants.fetch}-copy` }, + { kind: "orchestrator", name: `${BlobPurgeConstants.orchestrator}-copy` }, + { kind: "activity", name: BlobPurgeConstants.fetch.toLowerCase() }, + { kind: "orchestrator", name: BlobPurgeConstants.orchestrator.toLowerCase() }, + ]; + + async function dispatch( + kind: Kind, + name: string, + matchStrategy = VersionMatchStrategy.Strict, + failureStrategy = VersionFailureStrategy.Reject, + payloads = true, + version = "3.0", + ) { + const worker = new TaskHubGrpcWorker({ + payloads: payloads ? { store: new Store() } : undefined, + versioning: { version: "2.0", matchStrategy, failureStrategy }, + logger, + }); + const stub = new stubs.TaskHubSidecarServiceClient("127.0.0.1:1", grpc.credentials.createInsecure()); + worker["_stub"] = stub; + const execute = jest.fn(() => "done"); + const call = () => + Object.assign(new EventEmitter(), { + cancel: jest.fn(), + getPeer: () => "test", + getAuthContext: () => null, + }); + const activities: pb.ActivityResponse[] = []; + const orchestrations: pb.OrchestratorResponse[] = []; + const abandon = jest.fn(); + type Callback = (error: grpc.ServiceError | null, response: T) => void; + jest + .spyOn(stub, "completeActivityTask") + .mockImplementation( + ( + request, + _metadata, + optionsOrCallback: Partial | Callback, + callback?: Callback, + ) => { + activities.push(request); + (typeof optionsOrCallback === "function" ? optionsOrCallback : callback!)( + null, + new pb.CompleteTaskResponse(), + ); + return call(); + }, + ); + jest + .spyOn(stub, "completeOrchestratorTask") + .mockImplementation( + ( + request, + _metadata, + optionsOrCallback: Partial | Callback, + callback?: Callback, + ) => { + orchestrations.push(request); + (typeof optionsOrCallback === "function" ? optionsOrCallback : callback!)( + null, + new pb.CompleteTaskResponse(), + ); + return call(); + }, + ); + jest + .spyOn(stub, "abandonTaskActivityWorkItem") + .mockImplementation( + ( + _request, + _metadata, + optionsOrCallback: Partial | Callback, + callback?: Callback, + ) => { + abandon(); + (typeof optionsOrCallback === "function" ? optionsOrCallback : callback!)( + null, + new pb.AbandonActivityTaskResponse(), + ); + return call(); + }, + ); + jest + .spyOn(stub, "abandonTaskOrchestratorWorkItem") + .mockImplementation( + ( + _request, + _metadata, + optionsOrCallback: Partial | Callback, + callback?: Callback, + ) => { + abandon(); + (typeof optionsOrCallback === "function" ? optionsOrCallback : callback!)( + null, + new pb.AbandonOrchestrationTaskResponse(), + ); + return call(); + }, + ); + Object.defineProperty(stub, "makeUnaryRequest", { + value: ( + _path: string, + _serialize: (request: lp.GetLargePayloadTombstonesRequest) => Buffer, + _deserialize: (buffer: Buffer) => lp.GetLargePayloadTombstonesResponse, + _request: lp.GetLargePayloadTombstonesRequest, + _metadata: grpc.Metadata, + _options: grpc.CallOptions, + callback: Callback, + ) => { + callback(null, new lp.GetLargePayloadTombstonesResponse()); + return call(); + }, + }); + if (kind === "activity" && !worker["_registry"].getActivity(name)) worker.addNamedActivity(name, execute); + if (kind === "orchestrator" && !worker["_registry"].getOrchestrator(name)) { + worker.addNamedOrchestrator(name, async () => execute()); + } + const item = new pb.WorkItem().setCompletiontoken("kind-token"); + if (kind === "activity") { + item.setActivityrequest( + new pb.ActivityRequest() + .setName(name) + .setTaskid(1) + .setVersion(new StringValue().setValue(version)) + .setInput(new StringValue().setValue(name === BlobPurgeConstants.fetch ? "50" : "[]")) + .setOrchestrationinstance(new pb.OrchestrationInstance().setInstanceid("kind-test")), + ); + } else { + item.setOrchestratorrequest( + new pb.OrchestratorRequest() + .setInstanceid("kind-test") + .setNeweventsList([ + newOrchestratorStartedEvent(), + newExecutionStartedEvent( + name, + "kind-test", + '{"PurgeBatchSize":50,"PurgedCount":0}', + undefined, + undefined, + version, + ), + ]), + ); + } + try { + worker["_dispatchWorkItem"](item, stub); + await Promise.all(worker["_pendingWorkItems"]); + expect(worker["_pendingWorkItems"].size).toBe(0); + const failure = + kind === "activity" + ? activities[0]?.getFailuredetails() + : orchestrations[0] + ?.getActionsList() + .find((action) => action.hasCompleteorchestration()) + ?.getCompleteorchestration() + ?.getFailuredetails(); + return { execute, abandon, activities, orchestrations, failure }; + } finally { + stub.close(); + } + } + + describe.each([VersionMatchStrategy.Strict, VersionMatchStrategy.CurrentOrOlder])("strategy%s", (strategy) => { + it.each(customerTasks)("$kind $name keeps its mismatch policy", async ({ kind, name }) => { + const result = await dispatch(kind, name, strategy); + expect(result.execute).not.toHaveBeenCalled(); + expect(result.abandon).toHaveBeenCalledTimes(1); + expect(result.activities).toEqual([]); + expect(result.orchestrations).toEqual([]); + }); + it.each(wrongKind)("$kind $name keeps the Fail mismatch strategy", async ({ kind, name }) => { + const result = await dispatch(kind, name, strategy, VersionFailureStrategy.Fail); + expect(result.execute).not.toHaveBeenCalled(); + expect(result.abandon).not.toHaveBeenCalled(); + expect(result.failure?.getErrortype()).toBe("VersionMismatch"); + }); + }); + + it.each(reserved)("$kind $name is exempt only when it is a reserved task of that kind", async ({ kind, name }) => { + const result = await dispatch(kind, name, VersionMatchStrategy.Strict, VersionFailureStrategy.Fail); + expect(result.abandon).not.toHaveBeenCalled(); + expect(result.failure).toBeUndefined(); + expect(result.activities.length + result.orchestrations.length).toBe(1); + if (kind === "orchestrator") { + expect(result.orchestrations[0].getActionsList()[0].getScheduletask()?.getName()).toBe(BlobPurgeConstants.fetch); + } + }); + + it.each(reserved)("$kind $name has no exemption without payloads", async ({ kind, name }) => { + const result = await dispatch(kind, name, VersionMatchStrategy.Strict, VersionFailureStrategy.Reject, false); + expect(result.execute).not.toHaveBeenCalled(); + expect(result.abandon).toHaveBeenCalledTimes(1); + }); + + it.each(wrongKind)("$kind $name still runs when version matching is None", async ({ kind, name }) => { + const result = await dispatch(kind, name, VersionMatchStrategy.None); + expect(result.execute).toHaveBeenCalledTimes(1); + expect(result.abandon).not.toHaveBeenCalled(); + expect(result.failure).toBeUndefined(); + }); + + it.each(wrongKind)("$kind $name still accepts older versions under CurrentOrOlder", async ({ kind, name }) => { + const result = await dispatch( + kind, + name, + VersionMatchStrategy.CurrentOrOlder, + VersionFailureStrategy.Reject, + true, + "1.0", + ); + expect(result.execute).toHaveBeenCalledTimes(1); + expect(result.abandon).not.toHaveBeenCalled(); + expect(result.failure).toBeUndefined(); + }); +}); + +describe("purge delete classification", () => { + it("bounds deletes at eight and preserves positional dispositions without logging tokens", async () => { + const store = new Store(); + let active = 0; + let max = 0; + store.delete = async (token) => { + max = Math.max(max, ++active); + await new Promise((resolve) => setImmediate(resolve)); + active--; + if (token.endsWith("error")) throw Object.assign(new Error(token), { statusCode: 403 }); + return PayloadDeleteOutcome.Deleted; + }; + const tokens = Array.from({ length: 50 }, (_, i) => `blob:v2:https://a.invalid/c/${i === 3 ? "error" : i}`); + const warn = jest.fn(); + const outcomes = await deletePayloadChunk(store, tokens, { + ...logger, + warn, + error: jest.fn(), + info: jest.fn(), + debug: jest.fn(), + }); + expect(max).toBe(8); + expect(outcomes.map((o) => o.Disposition)).toEqual(tokens.map((_, i) => (i === 3 ? 2 : 1))); + expect(JSON.stringify(warn.mock.calls)).not.toContain("a.invalid"); + }); + it.each([ + ["blob:v1:c:b", 3], + ["blob:v3:future", 2], + ["blob:v2:not-a-url", 3], + ["blob:v2:", 3], + ["blob:v2:ftp://account.invalid/c/b", 3], + ["blob:v2:https://user:password@account.invalid/c/b", 3], + ["blob:v2:https://account.invalid/c/", 3], + ])("does not delete %s", async (token, disposition) => { + const store = new Store(); + store.delete = jest.fn(); + expect(await deletePayloadChunk(store, [token], logger)).toEqual([{ Disposition: disposition }]); + expect(store.delete).not.toHaveBeenCalled(); + }); + it.each([ + "blob:v2:https://account.invalid/c/nested/blob", + "blob:v2:https://account.invalid/c/folder/", + "blob:v2:https://account.invalid/c/%2F", + "blob:v2:http://127.0.0.1:10000/account/container/blob", + ])("preserves valid encoded, nested and path-style v2 tokens (%#)", async (token) => { + const store = new Store(); + store.delete = jest.fn(async () => PayloadDeleteOutcome.Deleted); + await expect(deletePayloadChunk(store, [token], logger)).resolves.toEqual([{ Disposition: 1 }]); + expect(store.delete).toHaveBeenCalledWith(token, expect.any(AbortSignal)); + }); + it.each([ + [new TypeError("uncertain store failure"), 2], + [Object.assign(new Error("bad"), { statusCode: 400 }), 3], + [Object.assign(new Error("bad"), { statusCode: 401 }), 2], + [Object.assign(new Error("bad"), { statusCode: 412 }), 2], + [new Error("network"), 2], + ])("classifies storage errors without losing tombstones (%#)", async (error, disposition) => { + const store = new Store(); + store.delete = async () => { + throw error; + }; + expect(await deletePayloadChunk(store, ["blob:v2:https://a.invalid/c/b"], logger)).toEqual([ + { Disposition: disposition }, + ]); + }); + + it("retries a native fetch connection failure on a valid v2 token without logging its URL or error", async () => { + const portHolder = createServer(); + await new Promise((resolve) => portHolder.listen(0, "127.0.0.1", resolve)); + const address = portHolder.address(); + if (!address || typeof address === "string") throw new Error("Expected an ephemeral loopback port"); + await new Promise((resolve, reject) => portHolder.close((error) => (error ? reject(error) : resolve()))); + const token = `blob:v2:http://127.0.0.1:${address.port}/c/b`; + // Native fetch errors must share the runtime realm with the classifier, not Jest's VM realm. + const script = ` + require("ts-node").register({project: process.argv[2], transpileOnly: true, swc: true}); + const {deletePayloadChunk} = require(process.argv[3]); + const controller = new AbortController(); + const warnings = []; + let observed, watchdogFired = false; + const store = {async delete(value, signal) { + signal.throwIfAborted(); + try { await fetch(value.slice("blob:v2:".length), {method: "DELETE", signal}); } + catch(error) { + observed = {name: error.name, causeCode: error.cause?.code, sameRealmTypeError: error instanceof TypeError}; + throw error; + } + }}; + const logger = {warn: value => warnings.push(value), info() {}, error() {}, debug() {}}; + const watchdog = setTimeout(() => {watchdogFired = true; controller.abort();}, 800); + deletePayloadChunk(store, [process.argv[1]], logger, controller.signal).then(outcomes => { + clearTimeout(watchdog); + console.log(JSON.stringify({outcomes, observed, warnings, watchdogFired, callerAborted: controller.signal.aborted})); + }, error => {clearTimeout(watchdog); console.error(error); process.exitCode = 1;}); + `; + const { stdout, stderr } = await promisify(execFile)( + process.execPath, + [ + "-e", + script, + token, + join(__dirname, "..", "tsconfig.json"), + join(__dirname, "..", "src", "payloads", "purge-worker.ts"), + ], + { cwd: join(__dirname, "../../.."), timeout: 4000 }, + ); + const result = JSON.parse(stdout); + expect(stderr).toBe(""); + expect(result.watchdogFired).toBe(false); + expect(result.observed).toEqual({ name: "TypeError", causeCode: "ECONNREFUSED", sameRealmTypeError: true }); + expect(result.outcomes).toEqual([{ Disposition: LargePayloadPurgeDisposition.Retry }]); + expect(result.callerAborted).toBe(false); + expect(JSON.stringify(result.warnings)).not.toContain("MalformedToken"); + expect(JSON.stringify(result.warnings)).not.toContain("127.0.0.1"); + expect(JSON.stringify(result.warnings)).not.toContain("fetch failed"); + }); + it.each([PayloadDeleteOutcome.Deleted, PayloadDeleteOutcome.AlreadyAbsent, PayloadDeleteOutcome.NotStoreOwned])( + "maps confirmed delete outcome %s to terminal success", + async (outcome) => { + const store = new Store(); + store.delete = async () => outcome; + expect(await deletePayloadChunk(store, ["blob:v2:https://a.invalid/c/b"], logger)).toEqual([{ Disposition: 1 }]); + }, + ); +}); + +describe("purge delete cancellation", () => { + const tokens = Array.from({ length: 50 }, (_, index) => `blob:v2:https://account.invalid/c/${index}`); + + it.each([{ input: [] }, { input: tokens }])( + "rejects pre-aborted chunks before invoking the store (%#)", + async ({ input }) => { + const controller = new AbortController(); + const reason = Object.freeze(new Error("Already canceled cleanup")); + const store = new Store(); + store.delete = jest.fn(async (_token: string, signal?: AbortSignal) => { + signal?.throwIfAborted(); + return PayloadDeleteOutcome.Deleted; + }); + controller.abort(reason); + await expect(deletePayloadChunk(store, input, logger, controller.signal)).rejects.toBe(reason); + expect(store.delete).not.toHaveBeenCalled(); + }, + ); + + it.each([new Error("Stop cleanup"), new TypeError("Storage-shaped cancellation")])( + "stops queued deletes, drains the active eight and preserves cancellation (%#)", + async (reason) => { + Object.freeze(reason); + const controller = new AbortController(); + const store = new Store(); + let markEight!: () => void; + const eightStarted = new Promise((resolve) => (markEight = resolve)); + let release!: () => void; + const drain = new Promise((resolve) => (release = resolve)); + const signals: AbortSignal[] = []; + let calls = 0; + let active = 0; + let peak = 0; + let settled = false; + let observed: unknown; + store.delete = async (_token, signal) => { + calls++; + if (!signal) throw new Error("Expected a delete signal"); + signal.throwIfAborted(); + signals.push(signal); + peak = Math.max(peak, ++active); + let onAbort = () => {}; + try { + return await new Promise((_resolve, reject) => { + onAbort = () => reject(signal.reason); + signal.addEventListener("abort", onAbort, { once: true }); + if (calls === 8) markEight(); + if (signal.aborted) onAbort(); + }); + } finally { + signal.removeEventListener("abort", onAbort); + await drain; + active--; + } + }; + const pending = deletePayloadChunk(store, tokens, logger, controller.signal).then( + () => { + settled = true; + }, + (error: unknown) => { + settled = true; + observed = error; + }, + ); + let watchdogFired = false; + const watchdog = setTimeout(() => { + watchdogFired = true; + controller.abort(reason); + release(); + }, 800); + try { + await Promise.race([ + eightStarted, + pending.then(() => { + throw new Error("Chunk settled before eight cancel-aware deletes started"); + }), + ]); + controller.abort(reason); + await new Promise((resolve) => setImmediate(resolve)); + expect(settled).toBe(false); + expect(calls).toBe(8); + release(); + await pending; + expect(watchdogFired).toBe(false); + expect(observed).toBe(reason); + expect(calls).toBe(8); + expect(peak).toBe(8); + expect(active).toBe(0); + expect(signals.every((signal) => getEventListeners(signal, "abort").length === 0)).toBe(true); + expect(getEventListeners(controller.signal, "abort")).toHaveLength(0); + } finally { + clearTimeout(watchdog); + controller.abort(reason); + release(); + await pending; + } + }, + ); + + it("does not turn a cancellation racing with a successful delete into success", async () => { + const controller = new AbortController(); + const reason = Object.freeze(new Error("Canceled as the delete completed")); + const store = new Store(); + store.delete = async () => { + controller.abort(reason); + return PayloadDeleteOutcome.Deleted; + }; + await expect(deletePayloadChunk(store, [tokens[0]], logger, controller.signal)).rejects.toBe(reason); + }); + + it("keeps the per-delete timeout retryable without canceling the caller", async () => { + jest.useFakeTimers(); + const controller = new AbortController(); + const store = new Store(); + let deleteSignal: AbortSignal | undefined; + store.delete = async (_token, signal) => { + deleteSignal = signal; + if (!signal) throw new Error("Expected a delete signal"); + signal.throwIfAborted(); + let onAbort = () => {}; + try { + return await new Promise((_resolve, reject) => { + onAbort = () => reject(signal.reason); + signal.addEventListener("abort", onAbort, { once: true }); + }); + } finally { + signal.removeEventListener("abort", onAbort); + } + }; + const pending = deletePayloadChunk(store, [tokens[0]], logger, controller.signal); + try { + await jest.advanceTimersByTimeAsync(59999); + expect(deleteSignal?.aborted).toBe(false); + await jest.advanceTimersByTimeAsync(1); + await expect(pending).resolves.toEqual([{ Disposition: LargePayloadPurgeDisposition.Retry }]); + expect(deleteSignal?.aborted).toBe(true); + expect(controller.signal.aborted).toBe(false); + expect(getEventListeners(deleteSignal!, "abort")).toHaveLength(0); + expect(jest.getTimerCount()).toBe(0); + } finally { + controller.abort(); + await pending.catch(() => {}); + jest.useRealTimers(); + } + }); +}); + +describe("replay-safe purge runner", () => { + it("chunks 50 tokens, schedules four chunks per wave, and echoes opaque tombstones unchanged", async () => { + const context = new RuntimeOrchestrationContext("job"); + const run = blobPurgeJobOrchestrator(context, { PurgeBatchSize: 500, PurgedCount: 2 }); + await run.next(); + const tombstones = Array.from({ length: 251 }, (_, i) => ({ + TombstoneToken: `opaque:${i}`, + PayloadToken: `blob:v2:https://a/c/${i}`, + })); + await run.next(tombstones); + let actions = Object.values(context._pendingActions); + expect(actions.filter((a) => a.getScheduletask()?.getName() === BlobPurgeConstants.delete)).toHaveLength(4); + await run.next(Array.from({ length: 4 }, () => Array.from({ length: 50 }, () => ({ Disposition: 1 })))); + actions = Object.values(context._pendingActions); + expect(actions.filter((a) => a.getScheduletask()?.getName() === BlobPurgeConstants.delete)).toHaveLength(6); + await run.next([Array.from({ length: 50 }, () => ({ Disposition: 1 })), [{ Disposition: 1 }]]); + const report = Object.values(context._pendingActions).find( + (a) => a.getScheduletask()?.getName() === BlobPurgeConstants.report, + )!; + expect(JSON.parse(report.getScheduletask()!.getInput()!.getValue())).toEqual( + tombstones.map((t) => ({ TombstoneToken: t.TombstoneToken, Disposition: 1 })), + ); + await run.return(); + }); + + it("drains delivered and buffered configuration before continue-as-new on cycle five", async () => { + const context = new RuntimeOrchestrationContext("job"); + const run = blobPurgeJobOrchestrator(context, { PurgeBatchSize: 500, PurgedCount: 3 }); + await run.next(); + for (let cycle = 0; cycle < 5; cycle++) { + await run.next([{ TombstoneToken: "opaque", PayloadToken: "blob:v2:https://a/c/b" }]); + await run.next([[{ Disposition: 1 }]]); + if (cycle === 4) { + context._pendingEvents.setbatchsize[0].complete(60); + context._receivedEvents.setbatchsize = [70, 80]; + } + await run.next(null); + } + expect(context._newInput).toEqual({ PurgeBatchSize: 80, PurgedCount: 8 }); + expect(context._saveEvents).toBe(true); + }); + + it("waits for explicit configuration on unsupported backends without scheduling a retry timer", async () => { + const context = new RuntimeOrchestrationContext("job"); + const call = jest.spyOn(context, "callActivity"); + const run = blobPurgeJobOrchestrator(context, { PurgeBatchSize: 500, PurgedCount: 0 }); + await run.next(); + const policy = call.mock.calls[0][2]!.retry; + expect(policy).toBeInstanceOf(RetryPolicy); + expect( + (policy as RetryPolicy).shouldRetry({ errorType: "System.NotImplementedException", message: "unsupported" }), + ).toBe(false); + const failure = new TaskFailedError( + "unsupported", + new pb.TaskFailureDetails().setErrortype("System.NotImplementedException"), + ); + const waiting = await run.throw(failure); + expect(waiting.value).toBe(context._pendingEvents.setbatchsize[0]); + expect(Object.values(context._pendingActions).some((a) => a.hasCreatetimer())).toBe(false); + expect(JSON.parse(context.getCustomStatus()!).Status).toBe("BackendUnsupported"); + context._pendingEvents.setbatchsize[0].complete(100); + await run.next(100); + expect(call).toHaveBeenCalledTimes(2); + await run.return(); + }); + + it("backs off all-Retry results and cancels its idle timer when configuration arrives", async () => { + const context = new RuntimeOrchestrationContext("job"); + const timer = jest.spyOn(context, "createTimer"); + const run = blobPurgeJobOrchestrator(context, { PurgeBatchSize: 10, PurgedCount: 0 }); + await run.next(); + await run.next([{ TombstoneToken: "opaque", PayloadToken: "blob:v2:https://a/c/b" }]); + await run.next([[{ Disposition: LargePayloadPurgeDisposition.Retry } satisfies BlobPurgeOutcome]]); + await run.next(null); + expect(timer).toHaveBeenCalledTimes(1); + expect((timer.mock.calls[0][0] as Date).getTime() - context.currentUtcDateTime.getTime()).toBe(60000); + context._pendingEvents.setbatchsize[0].complete(20); + await run.next(context._pendingEvents.setbatchsize[0]); + expect(timer.mock.results[0].value.isCanceled).toBe(true); + await run.return(); + }); + + it("waits for configuration before rolling over an unsupported fifth cycle", async () => { + const context = new RuntimeOrchestrationContext("job"); + const run = blobPurgeJobOrchestrator(context, { PurgeBatchSize: 500, PurgedCount: 0 }, logger); + await run.next(); + for (let cycle = 0; cycle < 4; cycle++) { + await run.next([{ TombstoneToken: "opaque", PayloadToken: "blob:v2:https://a/c/b" }]); + await run.next([[{ Disposition: 1 }]]); + await run.next(null); + } + const failure = new TaskFailedError( + "unsupported", + new pb.TaskFailureDetails().setErrortype("System.NotImplementedException"), + ); + const waiting = await run.throw(failure); + expect(waiting.done).toBe(false); + expect(context._newInput).toBeUndefined(); + expect(Object.values(context._pendingActions).some((action) => action.hasCreatetimer())).toBe(false); + context._pendingEvents.setbatchsize[0].complete(60); + context._receivedEvents.setbatchsize = [70]; + expect((await run.next(60)).done).toBe(true); + expect(context._newInput).toEqual({ PurgeBatchSize: 70, PurgedCount: 4 }); + expect(context._saveEvents).toBe(true); + }); +});