In the Node.js Web Streams API, awaiting transformStream.writable.getWriter().write() results in unexpected behavior.

[timeto-code]

App

Next.js + TS

Feature

A file upload feature allows uploading in chunks using streaming. Each chunk is saved to disk in the format chunk_1, chunk_2..., chunk_n. Once the client has sent all chunks, it sends a merge request. Upon receiving this request, the server performs the merge operation. During the merge process, the server sends progress updates via Server-Sent Events (SSE).

Issue

I'm using Node.js Web Streams API TransformStream to implement SSE for sending progress updates. The write() method of the transformStream.writable.getWriter().write(chunk) returns a promise, but when I await it, the loop seems to pause, as if it's blocked (without any errors), and the log message only prints once. If I don't await the write() method, everything works as expected, and the functionality runs smoothly. Why does awaiting the write() method cause this issue?

Client

const checkFileDone = async (fileName: string) => {
  /* ... */

  const searchParams = new URLSearchParams();
  searchParams.append('name', file.name);
  const eventSource = new EventSource(`/api/upload?${searchParams.toString()}`);

  eventSource.addEventListener('success', (event) => {
    /* ..... */
    eventSource!.close();
  });

  eventSource.addEventListener('failed', (event) => {
    /* ..... */
    eventSource!.close();
  });

  eventSource.onerror = (error) => {
    /* ..... */
    eventSource!.close();
  };
};

API

export const GET = async (req: NextRequest) => {
  const name = req.nextUrl.searchParams.get('name');

  if (!name) {
    return NextResponse.json(message('Invalid parameters', ERROR_CODE.API_UPLOAD_POST));
  }

  const chunksDir = path.join(directory.temp, name);

  let filehandle: fs.promises.FileHandle | null = null;
  let writeStream: fs.WriteStream | null = null;
  let transformWriter: WritableStreamDefaultWriter<Uint8Array> | null = null;

  try {
    const chunks = await fs.promises.readdir(chunksDir);

    const uniqueFilePath = generateUniquePath(directory.shared, name);
    filehandle = await fs.promises.open(uniqueFilePath, 'w');
    writeStream = filehandle.createWriteStream({
      highWaterMark: 1024 * 1024 * 50,
    });

    const transformStream = new TransformStream();
    transformWriter = transformStream.writable.getWriter();
    const encoder = new TextEncoder();

    for (let i = 0; i < chunks.length; i++) {
      console.log(`chunk_${i}`);

      const chunkPath = path.join(chunksDir, `chunk_${i}`);
      const chunk = await fs.promises.readFile(chunkPath);

      const status = writeStream.write(chunk);
      if (!status) {
        await new Promise((resolve) => writeStream!.once('drain', resolve));
      }

      /* Adding await will cause the code to block */
      await transformWriter.write(encoder.encode(`: heartbeat\n\n`));
    }

    return new NextResponse(transformStream.readable, {
      headers: {
        'Cache-Control': 'no-cache, no-transform',
        'Content-Type': 'text/event-stream; charset=utf-8',
      },
    });
  } catch (error) {
    /* ... */
  }
}

Guidelines

• I have read and understood this category's guidelines before making this post.

[madhuragrawal]

It seems like the issue is related to backpressure handling in the Web Streams API.

When you await transformStream.writable.getWriter().write(chunk), the await pauses until the stream is ready for more data, which can block subsequent chunks and interrupt real-time flow like SSE.

The reason your code blocks when awaiting transformWriter.write() in the loop has to do with the behavior of the WritableStreamDefaultWriter.write() method when used with Node.js streams, especially in the context of SSE. When you await each write call, you’re waiting for it to fully complete before proceeding to the next iteration, effectively serializing the loop in a way that prevents any further execution until the promise resolves. In this case, it causes the loop to stall as each write operation waits for the previous to finish.

If you need to manage backpressure (stream readiness), handle it with transformStream.writable.close() after all chunks are written.

[timeto-code]

Is it that transformWriter.write() behaves like writeStream.write(), where the write() action is only performed after the buffer is full, and then a promise is returned? I have set the chunk size to only 5MB. If this is the case, the issue described above could be explained by the fact that the size of a chunk does not meet the conditions necessary to trigger transformWriter.write(), which would prevent the promise from being returned and cause the code to block.

[Gesronet]

This issue with await transformStream.writable.getWriter().write() in the Node.js Web Streams API might stem from how the writable stream handles backpressure or the state of the stream. If the writable stream is not ready to accept more data, it could lead to unexpected behavior or errors.

To troubleshoot, ensure that you're properly managing the writable stream's state. You might want to check if the stream is closed or errored before writing. Additionally, consider using await writer.ready to wait for the stream to be ready before writing again. If you provide more details about the specific behavior you're encountering, I’d be happy to help further!