Uncaught errors during `_read()`

[mitchmcdee]

_read() is async but ReadableStream expects it to be synchronous (i.e. it is never awaited) which results in uncaught errors e.g. if this.reader.read() throws an error it will never be caught. This was happening for me when reader.read() was throwing network errors.

https://nodejs.org/dist/v18.0.0/docs/api/stream.html#errors-while-reading for additional context

[epicmet]

I think we can change it a little to fix it

import { Readable } from "node:stream";

// Until AWS-S3 sdk provide Node.js streams, we need to have a wrapper for it.
//
// Issue Link: https://github.com/aws/aws-sdk-js-v3/issues/5582
// Original Source of this class: https://github.com/Borewit/readable-web-to-node-stream
export class ReadableWebToNodeStream extends Readable {
  public bytesRead: number = 0;
  public released = false;

  private reader: ReadableStreamDefaultReader<any>;
  private pendingRead: Promise<any> | undefined;

  constructor(stream: ReadableStream) {
    super();
    this.reader = stream.getReader();
  }

  /**
   * @override
   *
   * @description
   * The `_read` method must not be async
   * It's guaranteed to be called again when `this.push` is done
   * @see https://nodejs.org/docs/latest-v20.x/api/stream.html#readable_readsize
   */
  public _read() {
    if (this.released) {
      this.push(null); // Signal EOF
      return;
    }

    this.pendingRead = this.reader
      .read()
      .then((data) => {
        delete this.pendingRead;
        if (data.done || this.released) {
          this.push(null); // Signal EOF
        } else {
          this.bytesRead += data.value.length;
          this.push(data.value); // Push new data to the queue
        }
      })
      .catch((err) => {
        this.destroy(err);
      });
  }

  /**
   * Cancel the stream ungracefully and immediately
   */
  public cancel() {
    this.reader.cancel()
  }

  /**
   * Close the stream gracefully after the last chunk has been emitted
   */
  public async close(): Promise<void> {
    await this.syncAndRelease();
  }

  private async waitForReadToComplete() {
    if (this.pendingRead) {
      await this.pendingRead;
    }
  }

  private async syncAndRelease() {
    this.released = true;
    await this.waitForReadToComplete();
    this.reader.releaseLock();
  }
}

[Borewit]

Sorry it took forever, but I included your change

[epicmet]

No worries. I should have made a pull request in the first place.
Thanks 🤝