diff --git a/packages/durabletask-js/src/client/client.ts b/packages/durabletask-js/src/client/client.ts index d0be02e..bb4fc68 100644 --- a/packages/durabletask-js/src/client/client.ts +++ b/packages/durabletask-js/src/client/client.ts @@ -954,11 +954,15 @@ export class TaskHubGrpcClient { stream.on("end", () => { stream.removeAllListeners(); + stream.on("error", () => {}); // Prevent unhandled "error" after cleanup + stream.destroy(); resolve(historyEvents); }); stream.on("error", (err: grpc.ServiceError) => { stream.removeAllListeners(); + stream.on("error", () => {}); // Prevent unhandled "error" after cleanup + stream.destroy(); // Return empty array for NOT_FOUND to be consistent with DTS behavior // (DTS returns empty stream for non-existent instances) and user-friendly if (err.code === grpc.status.NOT_FOUND) { diff --git a/packages/durabletask-js/test/client-stream-cleanup.spec.ts b/packages/durabletask-js/test/client-stream-cleanup.spec.ts new file mode 100644 index 0000000..d59330d --- /dev/null +++ b/packages/durabletask-js/test/client-stream-cleanup.spec.ts @@ -0,0 +1,161 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +/** + * Tests that TaskHubGrpcClient.getOrchestrationHistory properly cleans up the + * gRPC stream after completion or error. + * + * Without proper cleanup (removeAllListeners + no-op error handler + destroy), + * the stream can leak resources and a late "error" event after removeAllListeners() + * will be unhandled — crashing the Node.js process. + */ + +import { EventEmitter } from "events"; +import * as grpc from "@grpc/grpc-js"; +import { TaskHubGrpcClient } from "../src"; +import * as pb from "../src/proto/orchestrator_service_pb"; +import { Timestamp } from "google-protobuf/google/protobuf/timestamp_pb"; + +/** + * Creates a mock stream that tracks destroy and cancel calls. + */ +function createMockStream(): EventEmitter & { + destroy: jest.Mock; + cancel: jest.Mock; +} { + const mockStream = new EventEmitter() as EventEmitter & { + destroy: jest.Mock; + cancel: jest.Mock; + }; + mockStream.destroy = jest.fn(); + mockStream.cancel = jest.fn(); + return mockStream; +} + +/** + * Creates a mock gRPC ServiceError with the specified status code and details. + */ +function createGrpcError(code: grpc.status, details: string = ""): grpc.ServiceError { + const error = new Error(details) as grpc.ServiceError; + error.code = code; + error.details = details; + error.metadata = new grpc.Metadata(); + return error; +} + +/** + * Creates a HistoryChunk protobuf message with a single OrchestratorStarted event. + */ +function createHistoryChunk(): pb.HistoryChunk { + const chunk = new pb.HistoryChunk(); + const event = new pb.HistoryEvent(); + event.setEventid(1); + event.setTimestamp(Timestamp.fromDate(new Date())); + const orchestratorStarted = new pb.OrchestratorStartedEvent(); + event.setOrchestratorstarted(orchestratorStarted); + chunk.setEventsList([event]); + return chunk; +} + +/** + * Accesses the internal _stub on a TaskHubGrpcClient for test mocking. + */ +function getStub(client: TaskHubGrpcClient): any { + return (client as any)._stub; +} + +describe("TaskHubGrpcClient.getOrchestrationHistory stream cleanup", () => { + let client: TaskHubGrpcClient; + let mockStream: ReturnType; + + beforeEach(() => { + client = new TaskHubGrpcClient({ hostAddress: "localhost:4001" }); + mockStream = createMockStream(); + getStub(client).streamInstanceHistory = jest.fn().mockReturnValue(mockStream); + }); + + it("should destroy the stream after successful completion", async () => { + const historyPromise = client.getOrchestrationHistory("test-instance"); + + // Emit some data followed by end + mockStream.emit("data", createHistoryChunk()); + mockStream.emit("end"); + + const history = await historyPromise; + expect(history).toHaveLength(1); + expect(mockStream.destroy).toHaveBeenCalled(); + }); + + it("should destroy the stream after a gRPC INTERNAL error", async () => { + const historyPromise = client.getOrchestrationHistory("test-instance"); + + const grpcError = createGrpcError(grpc.status.INTERNAL, "Internal server error"); + mockStream.emit("error", grpcError); + + await expect(historyPromise).rejects.toThrow("An error occurred while retrieving the history"); + expect(mockStream.destroy).toHaveBeenCalled(); + }); + + it("should destroy the stream after NOT_FOUND error (resolves with empty array)", async () => { + const historyPromise = client.getOrchestrationHistory("test-instance"); + + const grpcError = createGrpcError(grpc.status.NOT_FOUND, "Not found"); + mockStream.emit("error", grpcError); + + const history = await historyPromise; + expect(history).toEqual([]); + expect(mockStream.destroy).toHaveBeenCalled(); + }); + + it("should destroy the stream after CANCELLED error", async () => { + const historyPromise = client.getOrchestrationHistory("test-instance"); + + const grpcError = createGrpcError(grpc.status.CANCELLED, "Cancelled"); + mockStream.emit("error", grpcError); + + await expect(historyPromise).rejects.toThrow("was canceled"); + expect(mockStream.destroy).toHaveBeenCalled(); + }); + + it("should not crash from a late error event after successful stream end", async () => { + const historyPromise = client.getOrchestrationHistory("test-instance"); + + mockStream.emit("end"); + await historyPromise; + + // After cleanup, a late error event must not throw. + // Without the no-op error handler added after removeAllListeners(), + // this would crash the process with an unhandled 'error' event. + expect(() => { + mockStream.emit("error", new Error("Late error after cleanup")); + }).not.toThrow(); + }); + + it("should not crash from a late error event after stream error", async () => { + const historyPromise = client.getOrchestrationHistory("test-instance"); + + const grpcError = createGrpcError(grpc.status.NOT_FOUND, "Not found"); + mockStream.emit("error", grpcError); + + await historyPromise; // Resolves with [] for NOT_FOUND + + // After cleanup, a late error event must not throw. + expect(() => { + mockStream.emit("error", new Error("Late error after cleanup")); + }).not.toThrow(); + }); + + it("should validate instanceId parameter", async () => { + await expect(client.getOrchestrationHistory("")).rejects.toThrow("instanceId is required"); + }); + + it("should rethrow unrecognized gRPC errors without wrapping", async () => { + const historyPromise = client.getOrchestrationHistory("test-instance"); + + const grpcError = createGrpcError(grpc.status.PERMISSION_DENIED, "Denied"); + mockStream.emit("error", grpcError); + + await expect(historyPromise).rejects.toBe(grpcError); + expect(mockStream.destroy).toHaveBeenCalled(); + }); +});