Checklist

• I have ensured my pull request is not behind the main or master branch of the original repository.
• I have rebased all commits where necessary so that reviewing this pull request can be done without having to merge it first.
• I have written a commit message that passes commitlint linting.
• I have ensured that my code changes pass linting tests.
• I have ensured that my code changes pass unit tests.
• I have described my pull request and the reasons for code changes along with context if necessary.

pipe(rs, res) will not destroy res, when an error occurs. Thus causing client "hang" infinite when it has received some part of data.

Following code example, a readable stream has been implemented. When client request, the server will send some data, but after the error occurs, the client hang, the server doesn't close the socket. Chrome will hang. curl and ab will end in timeout.

After replacing with pipeline, when error occur, koa immediately close the socket, Chrome will say "Check internet connection" in download page. curl and ab will finish request as soon as possible.

stream.pipeline() will call stream.destroy(err) on all streams except they have already been closed.

the behavior may be better than hanging the request. Instead of client waitting indefinitely for the remaining bytes, closing the connection will break the expectation, but it's the only way to signal that the response is incomplete. The client will know that the response is truncated. Headers are already sent, an error occurs mid-stream, the client could handle this predictably.

Here's another to consider:

It's koa duty to close connection, it's app writer's duty to close the readable stream, and it's (better) stream implementers duty to release corresponding resources(request socket, fd, etc) of the readable stream when error occurs(but if it didn't work well, app writer should handle it, not koa).

Last benifit is pipeline() added support for webstreams(v18.16.0), making it easier.

import { Readable } from 'node:stream';
import http from 'http';
import Koa from 'koa';

class MyStream extends Readable {
  #count = 0;
  _read(size) {
    this.#count++;
    this.push(':-)');
    if (this.#count === 4) {
      try {
        // https://nodejs.org/api/stream.html#errors-while-reading
        throw new Error('Error in stream');
      } catch (e) {
        this.destroy(e);
        return;
      }
    }
    if (this.#count === 5) {
      this.push(null);
    }
  }
}

const httpPort = 8081;
const app = new Koa();

app.use((ctx) => {
  if (ctx.request.path === '/favicon.ico') {
    ctx.status = 404;
    return;
  }

  const stream = new MyStream();

  ctx.body = stream;
});

http.createServer(app.callback()).listen(httpPort);