diff --git a/packages/durabletask-js-export-history/src/builders/use-export-history.ts b/packages/durabletask-js-export-history/src/builders/use-export-history.ts index af71914..8dfc4e1 100644 --- a/packages/durabletask-js-export-history/src/builders/use-export-history.ts +++ b/packages/durabletask-js-export-history/src/builders/use-export-history.ts @@ -1,7 +1,7 @@ // Copyright (c) Microsoft Corporation. All rights reserved. // Licensed under the MIT License. -import { TaskHubGrpcClient, TaskHubGrpcWorker } from "@microsoft/durabletask-js"; +import { TaskHubGrpcClient, TaskHubGrpcWorker, Logger } from "@microsoft/durabletask-js"; import { ExportHistoryStorageOptions } from "../models"; import { ExportHistoryClient } from "../client/export-history-client"; import { ExportJob } from "../entity/export-job"; @@ -97,6 +97,7 @@ export function useExportHistoryWorker( * * @param client The TaskHubGrpcClient to use for orchestration interactions. * @param storageOptions Azure Blob Storage configuration for the export destination. + * @param logger Optional logger for client-side diagnostics. Defaults to ConsoleLogger. * @returns A configured ExportHistoryClient instance. * * @example @@ -117,6 +118,7 @@ export function useExportHistoryWorker( export function createExportHistoryClient( client: TaskHubGrpcClient, storageOptions: ExportHistoryStorageOptions, + logger?: Logger, ): ExportHistoryClient { if (!storageOptions.connectionString) { throw new Error("ExportHistoryStorageOptions.connectionString is required"); @@ -126,5 +128,5 @@ export function createExportHistoryClient( throw new Error("ExportHistoryStorageOptions.containerName is required"); } - return new ExportHistoryClient(client, storageOptions); + return new ExportHistoryClient(client, storageOptions, logger); } diff --git a/packages/durabletask-js-export-history/src/client/export-history-client.ts b/packages/durabletask-js-export-history/src/client/export-history-client.ts index 9e13e84..2ee88cd 100644 --- a/packages/durabletask-js-export-history/src/client/export-history-client.ts +++ b/packages/durabletask-js-export-history/src/client/export-history-client.ts @@ -9,6 +9,8 @@ import { createAsyncPageable, Page, EntityQuery, + Logger, + ConsoleLogger, } from "@microsoft/durabletask-js"; import { ExportJobCreationOptions, @@ -36,13 +38,19 @@ import { ExportJobNotFoundError } from "../errors"; export class ExportHistoryClient { private readonly client: TaskHubGrpcClient; private readonly storageOptions: ExportHistoryStorageOptions; + private readonly logger: Logger; - constructor(client: TaskHubGrpcClient, storageOptions: ExportHistoryStorageOptions) { + constructor( + client: TaskHubGrpcClient, + storageOptions: ExportHistoryStorageOptions, + logger?: Logger, + ) { if (!client) throw new Error("client is required"); if (!storageOptions) throw new Error("storageOptions is required"); this.client = client; this.storageOptions = storageOptions; + this.logger = logger ?? new ConsoleLogger(); } /** @@ -120,7 +128,7 @@ export class ExportHistoryClient { */ getJobClient(jobId: string): ExportHistoryJobClient { if (!jobId) throw new Error("jobId is required"); - return new ExportHistoryJobClient(this.client, jobId, this.storageOptions); + return new ExportHistoryJobClient(this.client, jobId, this.storageOptions, this.logger); } } @@ -135,11 +143,13 @@ export class ExportHistoryJobClient { private readonly jobId: string; private readonly storageOptions: ExportHistoryStorageOptions; private readonly entityId: EntityInstanceId; + private readonly logger: Logger; constructor( client: TaskHubGrpcClient, jobId: string, storageOptions: ExportHistoryStorageOptions, + logger?: Logger, ) { if (!client) throw new Error("client is required"); if (!jobId) throw new Error("jobId is required"); @@ -149,6 +159,7 @@ export class ExportHistoryJobClient { this.jobId = jobId; this.storageOptions = storageOptions; this.entityId = new EntityInstanceId(EXPORT_JOB_ENTITY_NAME, jobId); + this.logger = logger ?? new ConsoleLogger(); } /** @@ -225,13 +236,32 @@ export class ExportHistoryJobClient { request, ); - // Then terminate the linked export orchestration if it exists + // Then terminate the linked export orchestration if it exists. + // Only swallow "not found" errors (orchestration doesn't exist or already purged). + // Re-throw all other errors (network failures, auth errors, timeouts, etc.). try { await this.client.terminateOrchestration(orchestrationInstanceId, "Export job deleted"); await this.client.waitForOrchestrationCompletion(orchestrationInstanceId, false, 30); await this.client.purgeOrchestration(orchestrationInstanceId); - } catch { - // Orchestration instance doesn't exist or already purged - this is expected + } catch (e: unknown) { + if (isNotFoundError(e)) { + // Expected: the linked orchestration doesn't exist or was already purged. + // Log at info level (aligned with the .NET SDK) but do not re-throw. + this.logger.info( + `Orchestration instance '${orchestrationInstanceId}' is already purged or never existed.`, + e, + ); + } else { + // Unexpected failure (network, auth, timeout, server error): log and re-throw + // so the caller is aware the cleanup did not complete. + this.logger.error( + `Failed to terminate or purge linked orchestration '${orchestrationInstanceId}': ${ + e instanceof Error ? e.message : String(e) + }`, + e, + ); + throw e; + } } } } @@ -268,3 +298,22 @@ function matchesFilter(state: ExportJobState, filter: ExportJobQuery): boolean { (state.createdAt !== undefined && new Date(state.createdAt) <= filter.createdTo); return statusMatches && createdFromMatches && createdToMatches; } + +/** gRPC status code for NOT_FOUND (well-known constant from the gRPC spec). */ +const GRPC_STATUS_NOT_FOUND = 5; + +/** + * Checks whether an error is a gRPC NOT_FOUND error. + * + * Uses duck typing to inspect the error's `code` property without requiring + * a direct dependency on `@grpc/grpc-js`. gRPC `ServiceError` objects always + * carry a numeric `code` field matching the standard gRPC status codes. + */ +function isNotFoundError(error: unknown): boolean { + return ( + typeof error === "object" && + error !== null && + "code" in error && + (error as { code: unknown }).code === GRPC_STATUS_NOT_FOUND + ); +} diff --git a/packages/durabletask-js-export-history/test/export-history-client-delete.spec.ts b/packages/durabletask-js-export-history/test/export-history-client-delete.spec.ts new file mode 100644 index 0000000..51e47a1 --- /dev/null +++ b/packages/durabletask-js-export-history/test/export-history-client-delete.spec.ts @@ -0,0 +1,213 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { ExportHistoryJobClient } from "../src/client/export-history-client"; +import { ExportHistoryStorageOptions } from "../src/models"; +import { ORCHESTRATOR_INSTANCE_ID_PREFIX } from "../src/constants"; +import { NoOpLogger, Logger } from "@microsoft/durabletask-js"; + +/** gRPC status code constants for test readability. */ +const GRPC_STATUS = { + NOT_FOUND: 5, + UNAVAILABLE: 14, + UNAUTHENTICATED: 16, + INTERNAL: 13, +}; + +/** + * Creates a mock gRPC error with the specified status code. + */ +function createGrpcError(code: number, message: string): Error & { code: number } { + const error = new Error(message) as Error & { code: number }; + error.code = code; + return error; +} + +/** + * Creates a minimal mock of TaskHubGrpcClient for testing the delete() method. + */ +function createMockClient(overrides?: { + scheduleNewOrchestration?: jest.Mock; + terminateOrchestration?: jest.Mock; + waitForOrchestrationCompletion?: jest.Mock; + purgeOrchestration?: jest.Mock; +}) { + return { + scheduleNewOrchestration: overrides?.scheduleNewOrchestration ?? jest.fn().mockResolvedValue("op-instance-123"), + terminateOrchestration: overrides?.terminateOrchestration ?? jest.fn().mockResolvedValue(undefined), + waitForOrchestrationCompletion: overrides?.waitForOrchestrationCompletion ?? jest.fn().mockResolvedValue(undefined), + purgeOrchestration: overrides?.purgeOrchestration ?? jest.fn().mockResolvedValue(undefined), + } as unknown as ConstructorParameters[0]; +} + +const TEST_STORAGE_OPTIONS: ExportHistoryStorageOptions = { + connectionString: "DefaultEndpointsProtocol=https;AccountName=test;AccountKey=key;EndpointSuffix=core.windows.net", + containerName: "test-container", +}; + +const TEST_JOB_ID = "test-job-1"; + +/** A logger that discards output, used to keep test output clean. */ +const silentLogger = new NoOpLogger(); + +/** Creates a logger whose methods are jest mocks, for asserting log calls. */ +function createSpyLogger(): jest.Mocked { + return { + error: jest.fn(), + warn: jest.fn(), + info: jest.fn(), + debug: jest.fn(), + }; +} + +describe("ExportHistoryJobClient.delete()", () => { + it("should complete successfully when orchestration exists and all cleanup succeeds", async () => { + const terminateMock = jest.fn().mockResolvedValue(undefined); + const waitMock = jest.fn().mockResolvedValue(undefined); + const purgeMock = jest.fn().mockResolvedValue(undefined); + + const client = createMockClient({ + terminateOrchestration: terminateMock, + waitForOrchestrationCompletion: waitMock, + purgeOrchestration: purgeMock, + }); + + const jobClient = new ExportHistoryJobClient(client, TEST_JOB_ID, TEST_STORAGE_OPTIONS, silentLogger); + await expect(jobClient.delete()).resolves.toBeUndefined(); + + const expectedOrchId = `${ORCHESTRATOR_INSTANCE_ID_PREFIX}${TEST_JOB_ID}`; + expect(terminateMock).toHaveBeenCalledWith(expectedOrchId, "Export job deleted"); + expect(waitMock).toHaveBeenCalledWith(expectedOrchId, false, 30); + expect(purgeMock).toHaveBeenCalledWith(expectedOrchId); + }); + + it("should swallow gRPC NOT_FOUND error from terminateOrchestration", async () => { + const terminateMock = jest.fn().mockRejectedValue( + createGrpcError(GRPC_STATUS.NOT_FOUND, "Orchestration not found"), + ); + + const client = createMockClient({ terminateOrchestration: terminateMock }); + const jobClient = new ExportHistoryJobClient(client, TEST_JOB_ID, TEST_STORAGE_OPTIONS, silentLogger); + + await expect(jobClient.delete()).resolves.toBeUndefined(); + }); + + it("should swallow gRPC NOT_FOUND error from purgeOrchestration", async () => { + const purgeMock = jest.fn().mockRejectedValue( + createGrpcError(GRPC_STATUS.NOT_FOUND, "Instance not found"), + ); + + const client = createMockClient({ purgeOrchestration: purgeMock }); + const jobClient = new ExportHistoryJobClient(client, TEST_JOB_ID, TEST_STORAGE_OPTIONS, silentLogger); + + await expect(jobClient.delete()).resolves.toBeUndefined(); + }); + + it("should re-throw gRPC UNAVAILABLE error (network failure)", async () => { + const terminateMock = jest.fn().mockRejectedValue( + createGrpcError(GRPC_STATUS.UNAVAILABLE, "Connection refused"), + ); + + const client = createMockClient({ terminateOrchestration: terminateMock }); + const jobClient = new ExportHistoryJobClient(client, TEST_JOB_ID, TEST_STORAGE_OPTIONS, silentLogger); + + await expect(jobClient.delete()).rejects.toThrow("Connection refused"); + }); + + it("should re-throw gRPC UNAUTHENTICATED error (auth failure)", async () => { + const terminateMock = jest.fn().mockRejectedValue( + createGrpcError(GRPC_STATUS.UNAUTHENTICATED, "Token expired"), + ); + + const client = createMockClient({ terminateOrchestration: terminateMock }); + const jobClient = new ExportHistoryJobClient(client, TEST_JOB_ID, TEST_STORAGE_OPTIONS, silentLogger); + + await expect(jobClient.delete()).rejects.toThrow("Token expired"); + }); + + it("should re-throw gRPC INTERNAL error (server failure)", async () => { + const purgeMock = jest.fn().mockRejectedValue( + createGrpcError(GRPC_STATUS.INTERNAL, "Internal server error"), + ); + + const client = createMockClient({ purgeOrchestration: purgeMock }); + const jobClient = new ExportHistoryJobClient(client, TEST_JOB_ID, TEST_STORAGE_OPTIONS, silentLogger); + + await expect(jobClient.delete()).rejects.toThrow("Internal server error"); + }); + + it("should re-throw timeout errors from waitForOrchestrationCompletion", async () => { + const waitMock = jest.fn().mockRejectedValue(new Error("Timed out waiting for orchestration")); + + const client = createMockClient({ waitForOrchestrationCompletion: waitMock }); + const jobClient = new ExportHistoryJobClient(client, TEST_JOB_ID, TEST_STORAGE_OPTIONS, silentLogger); + + await expect(jobClient.delete()).rejects.toThrow("Timed out waiting for orchestration"); + }); + + it("should re-throw plain errors without a gRPC code property", async () => { + const terminateMock = jest.fn().mockRejectedValue(new Error("Unexpected error")); + + const client = createMockClient({ terminateOrchestration: terminateMock }); + const jobClient = new ExportHistoryJobClient(client, TEST_JOB_ID, TEST_STORAGE_OPTIONS, silentLogger); + + await expect(jobClient.delete()).rejects.toThrow("Unexpected error"); + }); + + it("should still schedule entity deletion before attempting orchestration cleanup", async () => { + const scheduleMock = jest.fn().mockResolvedValue("op-instance-123"); + const terminateMock = jest.fn().mockRejectedValue( + createGrpcError(GRPC_STATUS.NOT_FOUND, "Orchestration not found"), + ); + + const client = createMockClient({ + scheduleNewOrchestration: scheduleMock, + terminateOrchestration: terminateMock, + }); + + const jobClient = new ExportHistoryJobClient(client, TEST_JOB_ID, TEST_STORAGE_OPTIONS, silentLogger); + await jobClient.delete(); + + expect(scheduleMock).toHaveBeenCalledTimes(1); + }); + + it("should log at info level (without re-throwing) when the orchestration is not found", async () => { + const logger = createSpyLogger(); + const terminateMock = jest.fn().mockRejectedValue( + createGrpcError(GRPC_STATUS.NOT_FOUND, "Orchestration not found"), + ); + + const client = createMockClient({ terminateOrchestration: terminateMock }); + const jobClient = new ExportHistoryJobClient(client, TEST_JOB_ID, TEST_STORAGE_OPTIONS, logger); + + await expect(jobClient.delete()).resolves.toBeUndefined(); + + const expectedOrchId = `${ORCHESTRATOR_INSTANCE_ID_PREFIX}${TEST_JOB_ID}`; + expect(logger.info).toHaveBeenCalledTimes(1); + expect(logger.info).toHaveBeenCalledWith( + expect.stringContaining(`Orchestration instance '${expectedOrchId}' is already purged or never existed`), + expect.anything(), + ); + expect(logger.error).not.toHaveBeenCalled(); + }); + + it("should log at error level and re-throw when cleanup fails with a non-not-found error", async () => { + const logger = createSpyLogger(); + const purgeMock = jest.fn().mockRejectedValue( + createGrpcError(GRPC_STATUS.INTERNAL, "Internal server error"), + ); + + const client = createMockClient({ purgeOrchestration: purgeMock }); + const jobClient = new ExportHistoryJobClient(client, TEST_JOB_ID, TEST_STORAGE_OPTIONS, logger); + + await expect(jobClient.delete()).rejects.toThrow("Internal server error"); + + const expectedOrchId = `${ORCHESTRATOR_INSTANCE_ID_PREFIX}${TEST_JOB_ID}`; + expect(logger.error).toHaveBeenCalledTimes(1); + expect(logger.error).toHaveBeenCalledWith( + expect.stringContaining(`Failed to terminate or purge linked orchestration '${expectedOrchId}'`), + expect.anything(), + ); + expect(logger.info).not.toHaveBeenCalled(); + }); +});