ReadableStream hangs on Windows named pipes when proxying file streams

#!/usr/bin/env node

import { createReadStream } from 'node:fs'
import { mkdir, rm, stat, writeFile } from 'node:fs/promises'
import { createServer } from 'node:http'
import { platform } from 'node:os'
import { dirname, join } from 'node:path'
import { fileURLToPath } from 'node:url'
import { Agent, fetch } from 'undici'

const __dirname = dirname(fileURLToPath(import.meta.url))
const TEST_DIR = join(__dirname, '.test')
const TEST_FILE = join(TEST_DIR, 'data.txt')

const isWindows = platform() === 'win32'
const socketPath = isWindows
  ? `\\\\.\\pipe\\test-${Date.now()}.sock`
  : `/tmp/test-${Date.now()}.sock`

const PORT = 3333

// Create ~2MB test file
async function createTestFile() {
  await mkdir(TEST_DIR, { recursive: true })
  const content = 'x'.repeat(2_000_000) // 2MB
  await writeFile(TEST_FILE, content)
  console.log(`Created ${content.length} byte test file\n`)
}

// Server that streams a file
const fileServer = createServer(async (req, res) => {
  const stats = await stat(TEST_FILE)
  res.writeHead(200, {
    'Content-Type': 'text/plain',
    'Content-Length': stats.size,
  })
  createReadStream(TEST_FILE).pipe(res)
})

const proxyUndici = createServer(async (req, res) => {
  console.log('[Proxy] Received request, fetching from file server via undici...')

  const agent = new Agent({ connect: { socketPath } })
  const response = await fetch('http://localhost/', { dispatcher: agent })

  console.log(`[Proxy] Got response: ${response.status} ${response.statusText}`)
  console.log(`[Proxy] Headers:`, Object.fromEntries(response.headers))

  res.writeHead(response.status, Object.fromEntries(response.headers))

  const reader = response.body.getReader()
  let chunks = 0
  let totalBytes = 0

  console.log('[Proxy] Starting to read body via Web Streams...')

  while (true) {
    console.log(`[Proxy] Reading chunk ${chunks + 1}...`)

    const result = await Promise.race([
      reader.read(),
      new Promise(resolve => setTimeout(() => resolve({ timeout: true }), 5000)),
    ])

    if (result.timeout) {
      console.error(`[Proxy] HUNG at chunk ${chunks + 1} after ${totalBytes} bytes - this is the bug!`)
      reader.releaseLock()
      break
    }

    const { done, value } = result
    if (done) {
      console.log(`[Proxy] Done reading. Total: ${chunks} chunks, ${totalBytes} bytes`)
      break
    }

    chunks++
    totalBytes += value.length
    console.log(`[Proxy] Chunk ${chunks}: ${value.length} bytes (total: ${totalBytes} bytes)`)
    res.write(value)
  }

  res.end()
  console.log('[Proxy] Response ended')
})

// Clean up socket file on Unix
async function cleanupSocket() {
  if (!isWindows) {
    try {
      await rm(socketPath, { force: true })
    }
    catch {}
  }
}

async function main() {
  await cleanupSocket()
  await createTestFile()

  // Start file server on named pipe
  await new Promise(resolve => fileServer.listen({ path: socketPath }, resolve))
  console.log('File server listening on pipe\n')

  console.log('Starting undici proxy...\n')
  await new Promise(resolve => proxyUndici.listen(PORT, resolve))

  console.log('[Client] Fetching from proxy...')
  const res = await fetch(`http://localhost:${PORT}/`)

  console.log(`[Client] Got response, reading body...`)
  const text = await res.text()
  console.log(`[Client] Success: ${text.length} bytes received\n`)

  proxyUndici.close()
  fileServer.close()
  await cleanupSocket()
  await rm(TEST_DIR, { recursive: true, force: true })
}

main().catch(console.error)