example node 16 short body closed socket bug

listener-pipe.mjs

import fs from 'fs';
import http from 'http';
import stream from 'stream';

class CountingStream extends stream.PassThrough {
  constructor(options) {
    super(options);
    this._bytes_in = 0;
    this._bytes_out = 0;
  }

  getProgress() {
    const progress = {
      bytes_in: this._bytes_in,
      bytes_out: this._bytes_out,
    };

    return progress;
  }

  _transform(data, encoding, callback) {
    if (encoding !== 'buffer') {
      data = Buffer.from(data, encoding);
    }
    this._bytes_in += data.length;
    this.emit('progress', this.getProgress());
    this.push(data);
    this._bytes_out += data.length;
    this.emit('progress', this.getProgress());
    return callback();
  }
}

const writer = (input, output) =>
  new Promise((resolve, reject) => {
    const pass_through = new CountingStream();
    let bytes = 0;
    output.on('close', () => resolve(bytes));
    input.on('error', (err) => reject(err));
    output.on('error', (err) => reject(err));
    pass_through.on('progress', ({bytes_out}) => (bytes += bytes_out));
    input.pipe(pass_through).pipe(output);
  });

const sleep = async function (ms) {
  return new Promise((resolve) => setTimeout(resolve, ms));
};

let i = 0;
const handler = async (req, res) => {
  const out = fs.createWriteStream(`/tmp/file-${i}.out`, 'utf-8');
  i++;
  // await sleep(10);
  const total = await writer(req, out);
  console.log('total:', total);
};

const server = http.createServer(handler);
server.listen(55555);

listener.mjs

import fs from 'fs';
import http from 'http';

const sleep = async function (ms) {
  return new Promise((resolve) => setTimeout(resolve, ms));
};

let i = 0;
const handler = async (req, res) => {
  const out = fs.createWriteStream(`/tmp/file-${i}.out`, 'utf-8');
  i++;
  await sleep(10);
  let total = 0;
  for await (const chunk of req.iterator()) {
    total += chunk.length;
    out.write(chunk);
  }
  out.end();
  res.end();
  console.log('total:', total);
};

const server = http.createServer(handler);
server.listen(55555);

putter.mjs

import {createConnection} from 'net';

const message = `PUT /stream/D2QB3HB4/stream.m3u8 HTTP/1.1\r\nTransfer-Encoding: chunked\r
User-Agent: ascope ffmpeg\r
Accept: */*\r
Connection: close\r
Host: localhost:55555\r
Icy-MetaData: 1\r
\r
88\r
#EXTM3U
#EXT-X-VERSION:3
#EXT-X-TARGETDURATION:4
#EXT-X-MEDIA-SEQUENCE:0
#EXT-X-PLAYLIST-TYPE:EVENT
#EXTINF:3.999000,
segment0000000.ts
\r
0\r
`;

const socket = createConnection(55555);
socket.end(message);