fetch() doesn't cancel ReadableStream request body when connection closes (ECONNRESET)

Why this is a good first issue

The issue involves complex stream handling and connection management.

AI Summary

The issue describes a problem where `fetch()` does not cancel a `ReadableStream` request body when the connection closes unexpectedly. This requires understanding and modifying stream and connection handling logic, which is complex and may involve edge cases.

Issue Description

Version

v24.13.0

Platform

Microsoft Windows NT 10.0.26200.0 x64 / Linux 6.6.87.2-microsoft-standard-WSL2 x86_64 x86_64

Subsystem

No response

What steps will reproduce the bug?

Receiver:

import http from "http"
import { Readable } from 'stream'


const server = http.createServer({
  highWaterMark: 1
}, async (req, res) => {
    console.log("Receiver connected")

    const writeStream = new WritableStream({
      start(_controller) {
        console.log('LOG (writeStream): Stream started')
      },
      write(value) {
        console.log('LOG (writeStream):', value.length, Date.now())
        // Simulate slow consumer
        return new Promise(res => setTimeout(res, 1000))
      },
      close() {
        console.log('LOG (writeStream): Stream closed')
      },
      abort(err) {
        console.error('LOG (writeStream): Stream aborted:', err);
      }
    }, {
      highWaterMark: 1
    })

    const readable = Readable.toWeb(
      req,
      { strategy: { highWaterMark: 1 } }
    ) as unknown as ReadableStream

    await readable
      .pipeTo(writeStream)
      .catch(console.error)

    console.log("Receiver done")
    return res.end("ok")
})

server.listen(8080, '0.0.0.0', () => console.log("Server listening on port 8080"))

Sender:

const CHUNK_SIZE = 1024 * 16
const TOTAL_CHUNKS = 1000000

console.log(`Uploading ${TOTAL_CHUNKS} chunks of ${CHUNK_SIZE} bytes`)

const inputStream = new ReadableStream({
  start() {
    this.i = 0
    console.log('LOG (inputStream): Stream started')
  },
  pull(controller) {
    if (this.i >= TOTAL_CHUNKS) {
      console.log('[Stream] Closing')
      return controller.close()
    }
    
    console.log('LOG (inputStream): pull() called - chunk', this.i)
    
    const chunk = new Uint8Array(CHUNK_SIZE)
    chunk.fill(this.i % 256)
    
    controller.enqueue(chunk)
    this.i++
  }
}, {
  highWaterMark: 1
})

const encoder = new TransformStream({
  async transform(ch