Skip to content

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

Open

Description

@ferrisxie

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

Activity

  1. ferrisxie commented on Mar 12, 2025

    @ferrisxie
    Author

    From my code it hangs on await reponse.text()
    Inside the undici code, it hangs on await reader.read() in '/lib/web/fetch/util.js'

    From the log and code above, we can get the timeline:
    2025-03-12 12:51:15,604: start fetch
    2025-03-12 12:51:15,613: get fetch reponse, start call await response.text()
    2025-03-12 12:51:15.912: await reader.read() returned 451 times, and current chunkSize is 2899968 (total size is 3032884)
    2025-03-12 12:51:16,127: abort signal called after 500ms, as expected
    hangs

    2025-03-12 12:51:30.913: returned by Promise.race in 15 seconds
    2025-03-12 12:51:30,916: print info todo with the reader, the reader promise status is still pendding.

    Image

  2. fwx5618177 commented on Mar 13, 2025

    @fwx5618177

    It looks like:

    1. Request state was not properly reset
    2. Queue processing was interrupted but not cleaned up
    3. The reader’s promise remains in a pending state
  3. fwx5618177 commented on Mar 24, 2025

    @fwx5618177

    I reproduce this issue by using express.

    Here are code:

    server.js

    const express = require('express')
    const crypto = require('crypto')
    
    // Function to create server instance
    function createServer(port) {
      const app = express()
      
      // Track request counts to selectively delay certain requests
      let requestCount = 0;
      
      // Global control variable to simulate server hanging
      let shouldStall = false;
      
      // Toggle the stall status every 30 seconds
      setInterval(() => {
        shouldStall = !shouldStall;
        console.log(`Server hang status changed to: ${shouldStall ? 'hanging' : 'normal'}`);
      }, 30000);
    
      app.get('/files/:size', (req, res) => {
        const sizeInKB = parseInt(req.params.size, 10)
        const sizeInBytes = sizeInKB * 1024
        
        requestCount++;
        
        // Set response headers
        res.setHeader('Content-Type', 'application/octet-stream')
        res.setHeader('Content-Length', sizeInBytes)
        
        // Set response timeout for each request to prevent complete connection interruption
        req.socket.setTimeout(300000); // 5 minute timeout
        
        // Prevent errors caused by client-side connection interruption
        req.on('close', () => {
          console.log(`Request #${requestCount}: Connection closed by client`);
        });
    
        // Generate random data
        const buffer = crypto.randomBytes(sizeInBytes)
        
        // Apply slow transfer strategy for all large file requests
        if (sizeInBytes > 100 * 1024) {  // Files larger than 100KB
          console.log(`Request #${requestCount}: Applying slow transfer (${sizeInKB}KB)`);
          
          // Use smaller chunks and more precise control
          const CHUNK_SIZE = 16 * 1024; // 16KB
          let offset = 0;
          let chunkCount = 0;
          
          // Progress counter
          const totalChunks = Math.ceil(sizeInBytes / CHUNK_SIZE);
          
          function sendNextChunk() {
            // Check if need to enter hanging state
            if (shouldStall && chunkCount > 1) {
              console.log(`Request #${requestCount}: Global hang signal detected, pausing data transfer [${chunkCount}/${totalChunks}]`);
              // Pause for a very long time, but don't end the response, keeping the client hanging
              setTimeout(sendNextChunk, 60000);  // Check again after 1 minute
              return;
            }
            
            if (offset >= sizeInBytes) {
              console.log(`Request #${requestCount}: Transfer completed`);
              res.end();
              return;
            }
            
            try {
              // Determine current chunk size
              const end = Math.min(offset + CHUNK_SIZE, sizeInBytes);
              const chunk = buffer.slice(offset, end);
              
              // Use callback to ensure data is written
              res.write(chunk, (err) => {
                if (err) {
                  console.error(`Request #${requestCount}: Write error`, err);
                  try { res.end(); } catch (e) {}
                  return;
                }
                
                offset += chunk.length;
                chunkCount++;
                
                // Determine delay based on chunk number
                let delay;
                
                // First two chunks sent quickly, subsequent chunks intentionally slowed
                if (chunkCount <= 2) {
                  delay = 50; // Send first two chunks quickly
                  console.log(`Request #${requestCount}: Sending fast chunk ${chunkCount}/${totalChunks}, size: ${chunk.length/1024}KB`);
                } else if (chunkCount === 3) {
                  // After the third chunk, enter extra long delay to ensure client reader.read() will hang
                  delay = 120000; // 2 minute delay
                  console.log(`Request #${requestCount}: Entering extra long delay ${delay}ms`);
                } else {
                  delay = 10000; // Subsequent chunks also slow
                  console.log(`Request #${requestCount}: Sending slow chunk ${chunkCount}/${totalChunks}, delay: ${delay}ms`);
                }
                
                // Set time to send the next chunk
                setTimeout(sendNextChunk, delay);
              });
            } catch (err) {
              console.error(`Request #${requestCount}: Send error`, err);
              try { res.end(); } catch (e) {}
            }
          }
          
          // Start sending
          sendNextChunk();
        } else {
          // Small files still have some delay to prevent all responses from completing too quickly
          setTimeout(() => {
            res.end(buffer);
          }, Math.random() * 2000); // 0-2000ms random delay
        }
      })
    
      return app.listen(port, () => {
        console.log(`Server running on port ${port} - http://localhost:${port}/files/:size`)
      })
    }
    
    // Start 4 server instances on different ports
    const ports = [3000, 3001, 3002, 3003]
    const servers = ports.map(port => createServer(port))
    
    // Add overall startup success log
    console.log('=================================================')
    console.log('✅ All servers started successfully!')
    console.log('=================================================')
    console.log(`🚀 Started ${ports.length} test server instances:`)
    ports.forEach(port => {
      console.log(`   - http://localhost:${port}`)
    })
    console.log('\n📄 Available endpoints:')
    console.log('   - GET /files/:size - Returns random data of specified size (KB)')
    console.log('     Example: http://localhost:3000/files/1024 will return 1MB data')
    console.log('\n⚙️  Press Ctrl+C to stop all servers')
    console.log('=================================================')
    
    // Add graceful shutdown handling
    process.on('SIGINT', () => {
      console.log('Shutting down servers...')
      servers.forEach(server => server.close())
      process.exit()
    })

    test.js

    // First load the async logger
    require('./asyncLogger');
    
    const undici = require('./index.js')
    
    // Use global async logger, fallback to console if it doesn't exist
    const logger = global.__asyncLogger || console;
    logger.info('Test started - Using logger:', global.__asyncLogger ? 'global.__asyncLogger' : 'console');
    
    async function downloadSingleFile(url, index) {
      const startTime = Date.now()
      const timeout = 5000; // Increased to 5 seconds
      logger.info(`[${index}] Starting request to ${url} at ${new Date().toISOString()}`)
    
      const controller = new AbortController()
      const reqTimer = setTimeout(() => {
        logger.info(`[${index}] Aborting request after 5000ms`)
        controller.abort()
      }, timeout)
    
      try {
        const response = await undici.fetch(url, {
          signal: controller.signal,
          headers: {},
          bodyTimeout: 500000, // Increased to 500 seconds to ensure no reading timeout interruption
          headersTimeout: 60000 // Increased header timeout
        })
    
        logger.info(`[${index}] Got response in ${Date.now() - startTime}ms, status: ${response.status}`)
    
        if (response.status >= 200 && response.status < 300) {
          logger.info(`[${index}] Starting to read body at ${new Date().toISOString()}`)
          const textStartTime = Date.now()
          
          // Add extra monitoring to detect long-running text() calls
          const textTimeout = setTimeout(() => {
            logger.warn(`[${index}] ⚠️ WARNING: response.text() has been running for 5 seconds without completing!`)
          }, 5000)
          
          // Additional: Pass request index to response object for use in readAllBytes
          // This is a special handling for undici's internal implementation
          if (response.body && response.body.getReader) {
            const originalGetReader = response.body.getReader.bind(response.body);
            response.body.getReader = function() {
              const reader = originalGetReader();
              // Inject index into reader object for tracking
              reader.__tempIndexIdx = index;
              
              // Additionally record read start time
              reader.__readStartTime = Date.now();
              
              // Add diagnostic log
              logger.info(`[${index}] Created reader instance, start time: ${new Date(reader.__readStartTime).toISOString()}`);
              
              return reader;
            };
          }
          
          // Increase timeout monitoring frequency
          const textTimeoutInterval = setInterval(() => {
            const elapsedTime = Date.now() - textStartTime;
            logger.warn(`[${index}] ⚠️ WARNING: response.text() has been running for ${elapsedTime/1000} seconds without completing!`);
          }, 3000);
          
          // This may get stuck, which is the issue we're trying to reproduce
          try {
            const text = await response.text();
            clearInterval(textTimeoutInterval);
            clearTimeout(textTimeout);
            
            const textDuration = Date.now() - textStartTime
            logger.info(`[${index}] Completed reading body in ${textDuration}ms, length: ${text.length}`)
            
            // Record longer response times
            if (textDuration > 2000) {
              logger.warn(`[${index}] ⚠️ Slow text() operation detected: ${textDuration}ms for ${url}`)
            }
            
            return { success: true, length: text.length, duration: textDuration }
          } catch (textError) {
            clearInterval(textTimeoutInterval);
            clearTimeout(textTimeout);
            logger.error(`[${index}] Text reading error:`, textError);
            throw textError;
          }
        } else {
          return { success: false, status: response.status }
        }
      } catch (err) {
        logger.error(`[${index}] Error:`, {
          name: err.name,
          message: err.message,
          cause: err.cause,
          timestamp: new Date().toISOString()
        })
        return { success: false, error: err.message }
      } finally {
        clearTimeout(reqTimer)
      }
    }
    
    async function runTest() {
      // Force requests for large files to increase probability of triggering the issue
      const totalRequests = 100 // Reduce number of requests, but each is a large file
      
      // Simulate 4 different hosts
      const hosts = [
        'http://localhost:3000',
        'http://localhost:3001',
        'http://localhost:3002',
        'http://localhost:3003'
      ]
      
      // Generate requests, all for large files
      const requests = Array.from({ length: totalRequests }, (_, i) => {
        // All requests for large files
        const size = Math.floor(Math.random() * 4000) + 2000; // 2MB-6MB
        
        const host = hosts[i % hosts.length]
        return `${host}/files/${size}`
      })
    
      // Use lower concurrency, focusing on fewer large files
      const concurrency = 10 // Reduce concurrency to allow server to process
      const results = []
      let completedRequests = 0
    
      // Implement more aggressive concurrency model
      const batches = []
      for (let i = 0; i < requests.length; i += concurrency) {
        batches.push(requests.slice(i, i + concurrency))
      }
    
      // Use concurrency pattern closer to the issue description
      logger.info(`Starting test with ${requests.length} total requests, concurrency: ${concurrency}`)
      logger.info(`Using ${hosts.length} different hosts`)
    
      // Add functionality to periodically record current status
      const statusInterval = setInterval(() => {
        logger.info(`Current progress: ${completedRequests}/${totalRequests} completed`);
        logMemoryUsage();
      }, 5000);
    
      for (let batchIndex = 0; batchIndex < batches.length; batchIndex++) {
        const batch = batches[batchIndex]
        const batchStartTime = Date.now()
    
        logger.info(`Starting batch ${batchIndex + 1}/${batches.length}, size: ${batch.length}`)
    
        // Start all requests simultaneously
        const batchPromises = batch.map((url, idx) => {
          const requestIndex = batchIndex * concurrency + idx
          return downloadSingleFile(url, requestIndex)
            .then(result => {
              completedRequests++
              if (completedRequests % 10 === 0) {
                logger.info(`Progress: ${completedRequests}/${totalRequests} completed`)
              }
              return result
            })
        })
    
        // Wait for all requests in this batch to complete
        const batchResults = await Promise.all(batchPromises)
        results.push(...batchResults)
    
        logger.info(`Completed batch ${batchIndex + 1}/${batches.length} in ${Date.now() - batchStartTime}ms`)
      }
    
      clearInterval(statusInterval);
      return results
    }
    
    // Add memory usage monitoring
    function logMemoryUsage() {
      const used = process.memoryUsage()
      logger.info('Memory usage:', {
        rss: `${Math.round(used.rss / 1024 / 1024)} MB`,
        heapTotal: `${Math.round(used.heapTotal / 1024 / 1024)} MB`,
        heapUsed: `${Math.round(used.heapUsed / 1024 / 1024)} MB`,
        external: `${Math.round(used.external / 1024 / 1024)} MB`
      })
    }
    
    // Record memory usage every 10 seconds
    const memoryInterval = setInterval(logMemoryUsage, 10000)
    
    runTest().then((results) => {
      clearInterval(memoryInterval)
      
      const successes = results.filter(r => r.success).length
      const failures = results.filter(r => !r.success).length
      
      // Calculate statistics for text() operations
      const textDurations = results
        .filter(r => r.success && r.duration)
        .map(r => r.duration)
      
      const avgDuration = textDurations.reduce((sum, d) => sum + d, 0) / (textDurations.length || 1)
      const maxDuration = Math.max(...(textDurations.length ? textDurations : [0]))
      const slowResponses = results.filter(r => r.success && r.duration > 2000).length
    
      console.log('\nTest Summary:')
      console.log(`Total requests: ${results.length}`)
      console.log(`Successful: ${successes}`)
      console.log(`Failed: ${failures}`)
      console.log(`Average text() duration: ${avgDuration.toFixed(2)}ms`)
      console.log(`Maximum text() duration: ${maxDuration}ms`)
      console.log(`Slow responses (>2000ms): ${slowResponses}`)
      
      logMemoryUsage()
    }).catch(console.error)

    asyncLogger.js

    const fs = require('fs');
    const path = require('path');
    const util = require('util');
    
    // Create log directory
    const LOG_DIR = path.join(__dirname, 'logs');
    if (!fs.existsSync(LOG_DIR)) {
      fs.mkdirSync(LOG_DIR);
    }
    
    // Log file path
    const LOG_FILE = path.join(LOG_DIR, `test-${new Date().toISOString().replace(/:/g, '-')}.log`);
    
    // Create write stream
    const logStream = fs.createWriteStream(LOG_FILE, { flags: 'a' });
    
    // Async logger
    class AsyncLogger {
      constructor() {
        this.queue = [];
        this.isProcessing = false;
        
        // Write logs from queue to file every second
        setInterval(() => this.processQueue(), 1000);
        
        // Ensure all logs are written when process exits
        process.on('exit', () => {
          this.processQueueSync();
        });
      }
    
      // Format log message
      formatMessage(level, message, ...args) {
        const timestamp = new Date().toISOString();
        const formattedArgs = args.map(arg => {
          if (typeof arg === 'object') {
            return util.inspect(arg, { depth: null });
          }
          return arg;
        }).join(' ');
        
        return `[${timestamp}] [${level}] ${message} ${formattedArgs}`.trim();
      }
    
      // Add log to queue
      log(level, message, ...args) {
        const formattedMessage = this.formatMessage(level, message, ...args);
        
        // Output to console
        if (level === 'ERROR') {
          console.error(formattedMessage);
        } else if (level === 'WARN') {
          console.warn(formattedMessage);
        } else {
          console.log(formattedMessage);
        }
        
        // Add to queue
        this.queue.push(formattedMessage);
        
        // Start processing if queue is not already being processed
        if (!this.isProcessing) {
          this.processQueue();
        }
      }
    
      // Process queue asynchronously
      async processQueue() {
        if (this.isProcessing || this.queue.length === 0) return;
        
        this.isProcessing = true;
        
        try {
          const messagesToProcess = [...this.queue];
          this.queue = [];
          
          // Write to file
          for (const message of messagesToProcess) {
            logStream.write(message + '\n');
          }
        } finally {
          this.isProcessing = false;
          
          // Continue processing if queue still has messages
          if (this.queue.length > 0) {
            setImmediate(() => this.processQueue());
          }
        }
      }
    
      // Process queue synchronously (used when process exits)
      processQueueSync() {
        if (this.queue.length === 0) return;
        
        const messagesToProcess = [...this.queue];
        this.queue = [];
        
        for (const message of messagesToProcess) {
          fs.appendFileSync(LOG_FILE, message + '\n');
        }
      }
    
      // Public methods
      info(message, ...args) {
        this.log('INFO', message, ...args);
      }
    
      warn(message, ...args) {
        this.log('WARN', message, ...args);
      }
    
      error(message, ...args) {
        this.log('ERROR', message, ...args);
      }
    
      debug(message, ...args) {
        this.log('DEBUG', message, ...args);
      }
    }
    
    // Export instance and set as global variable
    const logger = new AsyncLogger();
    global.__asyncLogger = logger;
    
    module.exports = logger;

    we need to update lib/web/fetch/util.js

    const sleep = (ms)=>{
      return new Promise((resolve) => setTimeout(() => resolve({value: false}), ms));
    }
    
    /**
     * @see https://streams.spec.whatwg.org/#readablestreamdefaultreader-read-all-bytes
     * @see https://streams.spec.whatwg.org/#read-loop
     * @param {ReadableStreamDefaultReader} reader
     */
    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}`)
        // Significantly reduce timeout time to ensure it triggers before reader.read()
        const { done, value: chunk } = await Promise.race([reader.read(), sleep(100)])
        timesInfo.push({
          time: new Date().toISOString(),
          times,
          length: byteLength
        })
        logger.info(`fetch enter readAllBytes with idx chunk: ${chunk === false}`)
        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 toString().includes('kType'))
            const kStreamSymbol = symbolKey.find(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){
              // Increase chance to detect pending status by increasing timeout to 500ms
              waitingReqStatus = await Promise.race([
                waitingReqPromimse.then(() => "fulfilled").catch(() => "rejected"),
                new Promise(resolve => setTimeout(() => resolve("pending"), 500))
              ]);
            }
            infoOfReader.waitingLength = waitingLength
            const _stream = _innerReader.stream
            const streamKeys = Object.getOwnPropertySymbols(_stream)
            const kStreamStatus = streamKeys.find(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.
      }
    }

    The root cause of the problem is:

    1. Server stalls after sending partial data
    • In our server simulation code, it intentionally adds a 120-second (2-minute) delay after sending the first few data chunks
    • This situation simulates a "half-open connection" problem in real-world scenarios, where the TCP connection remains open but data transfer is interrupted
    1. Read operation blocks indefinitely
    • reader.read() is a blocking operation that waits until new data arrives or the stream ends
    • But if the server neither sends new data nor closes the connection, this read operation will hang indefinitely
  4. fwx5618177 commented on Mar 24, 2025

    @fwx5618177

    Image

  5. mcollina commented on Mar 27, 2025

    @mcollina
    SponsorMember

    Can you put all the above in a repository and make it easy for use to reproduce?

  6. fwx5618177 commented on May 27, 2025

    @fwx5618177

    Can you put all the above in a repository and make it easy for use to reproduce?

    I have pushed these files on branch fix-4089, we could find files on reproduce-script.

    Directly run:

    1. node ./reproduce-script/server.js
    2. node ./reproduce-script/test.js
  7. mcollina commented on Jun 1, 2025

    @mcollina
    SponsorMember

    What repository?

  8. rene84 commented on Jun 18, 2025

    @rene84

    We also ran into an issue that very accurately resembles the issue reported here. We saw hangs on await response.text(), but only when the server was responding with 'transfer-encoding': 'chunked'.

    What may be relevant for this issue is: bumping to undici 7.10.0 solved it for us

  9. fwx5618177 commented on Jun 18, 2025

    @fwx5618177

    We also ran into an issue that very accurately resembles the issue reported here. We saw hangs on await response.text(), but only when the server was responding with 'transfer-encoding': 'chunked'.

    What may be relevant for this issue is: bumping to undici 7.10.0 solved it for us

    Yes, please read this repo, and I reproduced it. But I still have no idea to fix it.

    @mcollina here: https://github.com/fwx5618177/undici/tree/fix-4089/reproduce-script

  10. ferrisxie commented on Jun 19, 2025

    @ferrisxie
    Author

    hi, @rene84 ,as you said :

    only when the server was responding with 'transfer-encoding': 'chunked'.

    I have noticed that there is the same response header chunked in my case.

  11. pacocartones commented on Aug 9, 2026

    @pacocartones
    Contributor

    I was able to reproduce this deterministically and pin down the exact fix.

    Root cause (undici 6.x). The hang is the same bug that #3754 fixed on the 7.x line (a regression introduced by 49e33a8 / #2823). In 6.21.1, when the abort lands after the transport has completed the body but before the body stream is closed:

    1. onComplete removes the terminated listener (lib/web/fetch/index.js, the off('terminated', fetchParams.controller.onAborted) added by remove all fetchParam event handlers #2823).
    2. The abort then emits terminated, but onAborted is gone, so fetchParams.controller.controller.error(...) never runs — the body stream is never errored.
    3. The resume loop wakes up, sees isAborted(fetchParams) and breaks without closing or erroring the stream.
    4. The ReadableStream stays 'readable' forever, so readAllBytes parks at await reader.read() in lib/web/fetch/util.js and response.text() never settles — exactly what @ferrisxie traced.

    Verified by instrumenting 6.21.1: onAborted is registered, onComplete removes it, the loop breaks on isAborted, and onAborted is never called. The read promise stays pending indefinitely.

    Repro. The deterministic repro is essentially the regression test added in #3754 (test/fetch/issue-1711.js, "aborting a Stream throws"): server sends a small full body immediately, client starts response.arrayBuffer() and aborts right away. On 6.21.1 this hangs 50/50 runs (Windows, node 24.14.1; also 20/20 on Linux node 18.20.8). On 7.10.0 and on current main, 50/50 reject with AbortError in ~1 ms.

    Note: the reproduce-script in https://github.com/fwx5618177/undici/tree/fix-4089/reproduce-script stalls the server mid-body, so the transport never completes and the abort path works (text() rejects with AbortError) — at least on node 24, I could not get it to hang across several 100-request runs. The hang window requires the abort to land after onComplete (body fully received at the socket level, still being drained), which matches the original timeline (451 reads returned, abort at 500 ms, then hang).

    Version bisect. 7.0.0-alpha.3 hangs, 7.0.0-alpha.4 is fixed — the only fetch/abort change in that range is afeb626 (#3754). First stable release with the fix: 7.0.0. This also explains @rene84's report that 7.10.0 fixed it.

    Still affected: the whole 6.x line — I verified 6.21.1, 6.21.3 and 6.28.0 (latest 6.x) all hang. The fix was never backported.

    So: fixed in 7.x by afeb626 (#3754). Two options: backport #3754 to the 6.x branch for users pinned to 6.x, or close this as fixed in 7.x with a note that 6.x won't get the fix.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    bugSomething isn't working

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions