Hangs on `await response.text()` using AbortController on unidci 6.21.1

Bug Description

I am upgrading my project from undici 5.x to 6.21.1. After the upgrade, I found that under high concurrency, when using undici.fetch to download files(The file sizes range from 10KB to 10MB, with concurrency typically between 2 and 10 requests. The total number of files is over 200, involving 4 different hosts.), there is a chance that it will block on await response.text(). I can’t consistently reproduce the issue, but I have added a lot of logs and gathered some information bellow. I would like to ask if, logically, there are cases where [await reponse.text()] does not return in some race condition?

Reproducible By

Our project is a Node.js web service that downloads many files during startup. This issue occurs when downloading the files at startup. The project code is quite complex, so I wrote a minimal demo for the downloading part. The only parts related to Undici and the APIs used are as follows. (We’re not sure if any of the third-party packages we depend on use Undici, as Node.js has Undici built-in, but the issue happens within the logic below.)

By the way, it cannot be reliably reproduced, and every time the issue occurs on different file, not the same file.

const downloadSingleFile = async (urlToDownload)=>{
    const options = {
        url: urlToDownload,
        method: 'GET',
        headers: {}
    }
    const controller = new AbortController();
    const timeout = 500
    const reqTimer = setTimeout(() => {
      logger.info(`fetch abort by timeout ${timeout}`);
      controller.abort();
    }, timeout);
    options.signal = controller.signal;
    let result;
    try{
        const response = await undici.fetch(options.url, options)
        const { status, statusText, headers } = response
        if (status >= 200 && status < 300) {
            const bodyTxt = await response.text();
            result = { statusCode: status, headers, body: bodyTxt };
        } else {
            result = { statusCode: status, headers, body: statusText };
        }
    }catch(e){
        result = {statusCode: -100, body: e.message}
    }
    clearTimeout(reqTimer);
    return result;
}
const demoEntry = async()=>{
    const res = [
        'https://xxxxxx/xxx.json'
        //...
        //about 200-300 urls
    ]
    for (let index = 0; index < res.length; index++) {
        const element = res[index];
        //...
        //concurrency is 1-10, not all urls
        //file size is between 10KB-10MB
        downloadSingleFile(element,index)
    }
}
demoEntry()

Expected Behavior

All promise return normally.

Logs & Screenshots

As the code shown above, I've add a lot of logs on response.text(), as it's hanged here.
The log i added is readAllBytes in /node_modules/undici/lib/web/fetch/util.js

//in my code
const controller = new AbortController();
const reqTimer = setTimeout(() => {
  logger.info(`fetch abort by timeout ${timeout} tempIndex=${tempIndex}`);
  controller.abort();
}, timeout);

//in /node_modules/undici/lib/web/fetch/util.js
const sleep = (ms)=>{
  return new Promise((resolve) => setTimeout(() => resolve({value: false}), ms));
}

async function readAllBytes (reader) {
  const bytes = []
  let byteLength = 0
  const logger = global.__asyncLogger;
  const tempIndexIdx = reader["__tempIndexIdx"] //this is injected from the download index
  logger.info(`fetch enter readAllBytes with idx ${tempIndexIdx} step0 ${!!reader}`)
  let times = 0
  let timesInfo = []
  while (true) {
    times++;
    // logger.info(`fetch enter readAllBytes with dix ${tempIndexIdx} step0 times=${times}`)
    const { done, value: chunk } = await Promise.race([reader.read(), sleep(15000)])
    timesInfo.push({
      time: new Date().toISOString(),
      times,
      length: byteLength
    })
    if(chunk === false){
      try{
        logger.info(`fetch enter controller readAllBytes with idx ${tempIndexIdx} step3 timeout ${!!reader} times=${times} length=${byteLength} ${JSON.stringify(timesInfo)}`)
        logger.info(`fetch enter readAllBytes with idx ${tempIndexIdx} step3 info 1`, {reader})
        const infoOfReader = {}
        const symbolKey = Object.getOwnPropertySymbols(reader)
        const kStateSymbol = symbolKey.find(sym => sym.toString().includes('kType'))
        const kStreamSymbol = symbolKey.find(sym => sym.toString().includes('kState'))
        const _innerState = reader[kStateSymbol]
        const _innerReader = reader[kStreamSymbol]
        infoOfReader.state = _innerState
        infoOfReader.state = _innerState
        const waitingLength = _innerReader.readRequests?.length || -1
        const waitingReq = _innerReader.readRequests[0]
        const waitingReqPromimse = waitingReq?.promise
        let waitingReqStatus = 'none';
        if(waitingReqPromimse){
          waitingReqStatus = await Promise.race([
            waitingReqPromimse.then(() => "fulfilled").catch(() => "rejected"),
            new Promise(resolve => setTimeout(() => resolve("pending"), 0))
          ]);
        }
        infoOfReader.waitingLength = waitingLength
        const _stream = _innerReader.stream
        const streamKeys = Object.getOwnPropertySymbols(_stream)
        const kStreamStatus = streamKeys.find(sym => sym.toString().includes('kState'))
        const stateOfStream = _stream[kStreamStatus]
        infoOfReader.stateOfStream = stateOfStream
        const controller = stateOfStream.controller
        const stateOfController = controller[kStreamStatus]
        infoOfReader.controllerState = stateOfController
        infoOfReader.waitingReqStatus = waitingReqStatus
        logger.info(`fetch enter readAllBytes with idx ${tempIndexIdx} status=${waitingReqStatus} step3 info ${JSON.stringify(infoOfReader)}`)
      }catch(e){
        logger.info(`fetch enter readAllBytes with idx ${tempIndexIdx} step3 timeout error ${e.message}`, e)
      }
     //when hangs happened, it will await here, so I can catch the hang.
      await sleep(600000)
    }
    if (done) {
      logger.info(`fetch enter readAllBytes with idx ${tempIndexIdx} step2 done`)
      // 1. Call successSteps with bytes.
      return Buffer.concat(bytes, byteLength)
    }

    // 1. If chunk is not a Uint8Array object, call failureSteps
    //    with a TypeError and abort these steps.
    if (!isUint8Array(chunk)) {
      throw new TypeError('Received non-Uint8Array chunk')
    }

    // 2. Append the bytes represented by chunk to bytes.
    bytes.push(chunk)
    byteLength += chunk.length

    // 3. Read-loop given reader, bytes, successSteps, and failureSteps.
  }
}

Here is useful log I've got when hangs.

2025-03-12 12:51:15,604 downloadIndex=340 startFetch
2025-03-12 12:51:15,613 downloadIndex=340 get fetch response, start await reponse.text()
2025-03-12 12:51:15,613 fetch enter readAllBytes with idx 340 step0 true
2025-03-12 12:51:16,127 fetch abort by timeout 500 tempIndex=340
2025-03-12 12:51:30,913 fetch enter controller readAllBytes with idx 340 step3 timeout true times=452 length=2899968 [...(ignores),{"time":"2025-03-12T12:51:15.912Z","times":451,"length":2890356},{"time":"2025-03-12T12:51:30.913Z","times":452,"length":2899968}]
2025-03-12 12:51:30,915 fetch enter readAllBytes with idx 340 step3 info 1 data: {
    reader: ReadableStreamDefaultReader {
      stream: ReadableStream { locked: true, state: 'readable', supportsBYOB: true },
      readRequests: 1,
      close: Promise {
        <pending>,
        [Symbol(async_id_symbol)]: 1145919,
        [Symbol(trigger_async_id_symbol)]: 1144674,
        [Symbol(kResourceStore)]: [Object]
      }
    }
  }
}
2025-03-12 12:51:30,916 INFO 128 [agent][node@v20.11.1] [ssr:1.202.0]  fetch enter readAllBytes with idx 340 status=pending step3 info {"state":"ReadableStreamDefaultReader","waitingLength":1,"stateOfStream":{"disturbed":true,"reader":{"__tempIndexIdx":340},"state":"readable","transfer":{},"controller":{}},"controllerState":{"byobRequest":null,"closeRequested":false,"pullAgain":false,"pulling":false,"started":true,"stream":{},"queue":[],"queueTotalSize":0,"highWaterMark":0,"pendingPullIntos":[]},"waitingReqStatus":"pending"}

As we can see the promise in the reader is still pennding, but the chunk is fully pulled and the abortController.sigal has emitted.

Environment

Debian GNU/Linux 12
tested on Node v20.11.1 and v22.12.0

Start at lib/web/fetch/util.js and its readAllBytes loop, then trace response.text() and AbortController through the supplied downloadSingleFile demo on Node v20 or v22. Reproduce or instrument the pending reader state under concurrent downloads; done means the cause of the hang is explained and all promises return normally after completion or abort.