Uncaught Exception on `node:stream/promises` pipeline

[felixjb]

Version

v22.16.0

Platform

Darwin Felixs-MacBook-Pro.local 24.5.0 Darwin Kernel Version 24.5.0: Tue Apr 22 19:54:29 PDT 2025; root:xnu-11417.121.6~2/RELEASE_ARM64_T6030 arm64

Subsystem

No response

What steps will reproduce the bug?

Given the following TypeScript function:

// Some details omitted

import { Transform } from "node:stream";
import { pipeline } from "node:stream/promises";

export const exportDataToCsvFile = async (
  dataStream: NodeJS.ReadableStream,
): Promise<void> => {
  const dataTransform = new Transform({
    objectMode: true,
    transform: (
      documentSnapshot: QueryDocumentSnapshot,
      _encoding,
      callback,
    ): void => {
      try {
        callback(null, documentSnapshot.data());
      } catch (error: unknown) {
        const errorInstance =
          error instanceof Error ? error : new Error(JSON.stringify(error));
        callback(errorInstance);
      }
    },
  });

  const csvStringifier = csv.stringify({ header: true });

  const filePath = storageService.buildFilePath({
    storeId: job.storeId,
    directory: job.name,
    name: `${getTimestamp()}.csv`,
  });
  const uploadStream = storageService.createUploadStream(filePath, {
    contentType: "text/csv",
  });

  try {
    await pipeline(dataStream, dataTransform, csvStringifier, uploadStream);
  } catch (error: unknown) {
    console.error("Failed to export data to CSV file", error);
  }
};

And the following Vitest test:

  it("should handle stream failures gracefully", async () => {
    // arrange - omitted 

    const failingDataStream = new Readable({
      read(): void {
        this.push("Not a document snapshot");
      },
    });

    await exportDataToCsvFile(failingDataStream);

    // assertions - omitted

The assertions pass and the test pass, at first, and after the test is finished an Unhandled Exception causes the test to fail:

⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯ Unhandled Errors ⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯

Vitest caught 1 unhandled error during the test run.
This might cause false positive tests. Resolve unhandled errors to make sure your tests are not affected.

⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯ Uncaught Exception ⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯
TypeError: documentSnapshot.data is not a function
 ❯ Transform.transform [as _transform] src/jobs/application/exportDataToCsvFile.ts:36:72
     34|     ): void {
     35|       try {
     36|         callback(null, documentSnapshot.data());
       |                                                                        ^
     37|       } catch (error: unknown) {
 ❯ Transform._write node:internal/streams/transform:171:8
 ❯ writeOrBuffer node:internal/streams/writable:572:12
 ❯ _write node:internal/streams/writable:501:10
 ❯ Transform.Writable.write node:internal/streams/writable:510:10
 ❯ Readable.ondata node:internal/streams/readable:1009:22
 ❯ Readable.emit node:events:518:28
 ❯ Readable.read node:internal/streams/readable:782:10
 ❯ flow node:internal/streams/readable:1283:53
 ❯ resume_ node:internal/streams/readable:1262:3

This error originated in "test/integrations/jobs/exportDataToCsvFile.test.ts" test file. It doesn't mean the error was thrown inside the file itself, but while it was running.
The latest test that might've caused the error is "should return execution failure and set job status to FAILED when the pipeline fails". It might mean one of the following:
- The error was thrown, while Vitest was running this test.
- If the error occurred after the test had been completed, this was the last documented test before it was thrown.
⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯


 Test Files  1 passed (1)
      Tests  1 passed | 3 skipped (4)
     Errors  1 error
   Start at  12:51:17
   Duration  1.41s (transform 524ms, setup 0ms, collect 1.15s, tests 143ms, environment 0ms, prepare 33ms)

How often does it reproduce? Is there a required condition?

It happens regardless of how many .catch() or .on('error') are in place. It also does not respond to ending or destroying the streams.

What is the expected behavior? Why is that the expected behavior?

I would like the pipeline to end when an error happens, and for all errors to be properly caught in the .catch() or try/catch.

What do you see instead?

Unhandled error

Additional information

No response