perf(h1): drop idle-socket timer floor with a ref'd setImmediate (#5707) (#5769)

(cherry picked from commit 8c9182e)

Signed-off-by: Matteo Collina <hello@matteocollina.com>
Co-authored-by: Yagiz Nizipli <yagiz@nizipli.com>

Author: mcollina

benchmarks/fetch/sequential-keepalive.mjs

@@ -0,0 +1,48 @@
+'use strict'
+
+import { createServer } from 'node:http'
+import { Agent, fetch } from '../../index.js'
+
+const ITERATIONS = Number(process.env.SAMPLES ?? 2000)
+const WARMUP = 300
+
+const server = createServer((req, res) => {
+  res.writeHead(200, { 'content-type': 'application/json' })
+  res.end('{"ok":1}')
+})
+
+server.keepAliveTimeout = 65_000
+
+await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve))
+
+const { port } = server.address()
+const url = `http://127.0.0.1:${port}/`
+const agent = new Agent({
+  keepAliveTimeout: 60_000,
+  connections: 1,
+  pipelining: 1
+})
+
+for (let i = 0; i < WARMUP; i++) {
+  await (await fetch(url, { dispatcher: agent })).text()
+}
+
+const times = new Array(ITERATIONS)
+for (let i = 0; i < ITERATIONS; i++) {
+  const t0 = process.hrtime.bigint()
+  await (await fetch(url, { dispatcher: agent })).text()
+  times[i] = Number(process.hrtime.bigint() - t0)
+}
+
+times.sort((a, b) => a - b)
+const pct = (p) => times[Math.min(ITERATIONS - 1, Math.floor(ITERATIONS * p))] / 1e6
+
+console.log(JSON.stringify({
+  iterations: ITERATIONS,
+  p50_ms: Number(pct(0.5).toFixed(3)),
+  p90_ms: Number(pct(0.9).toFixed(3)),
+  p99_ms: Number(pct(0.99).toFixed(3))
+}))
+
+await agent.close()
+server.close()

lib/dispatcher/client-h1.js

@@ -1012,7 +1012,7 @@ function onSocketClose () {
 
 function clearIdleSocketValidation (socket) {
   if (socket[kIdleSocketValidationTimeout]) {
-    clearTimeout(socket[kIdleSocketValidationTimeout])
+    clearImmediate(socket[kIdleSocketValidationTimeout])
     socket[kIdleSocketValidationTimeout] = null
   }
 
@@ -1021,15 +1021,23 @@ function clearIdleSocketValidation (socket) {
 
 function scheduleIdleSocketValidation (client, socket) {
   socket[kIdleSocketValidation] = 1
-  socket[kIdleSocketValidationTimeout] = setTimeout(() => {
+  // Yield to the check phase (after poll) so unsolicited bytes / FIN / RST
+  // already pending on this idle keep-alive socket are processed before the
+  // next request is written (GHSA-35p6-xmwp-9g52).
+  //
+  // setTimeout(0) pays Node's ~1ms timer floor on every sequential reuse
+  // (#5493). setImmediate avoids that, but an *unref'd* Immediate lets poll
+  // block for ~500ms when the event loop is otherwise idle (#5600 / #5606).
+  // A ref'd Immediate both keeps the pending request alive and makes poll
+  // return immediately — the hybrid those issues asked for.
+  socket[kIdleSocketValidationTimeout] = setImmediate(() => {
     socket[kIdleSocketValidationTimeout] = null
     socket[kIdleSocketValidation] = 2
 
     if (client[kSocket] === socket && !socket.destroyed) {
       client[kResume]()
     }
-  }, 0)
-  socket[kIdleSocketValidationTimeout].unref?.()
+  })
 }
 
 /**

test/node-test/keep-alive-reuse.js

@@ -0,0 +1,114 @@
+'use strict'
+
+const { test } = require('node:test')
+const assert = require('node:assert')
+const { createServer } = require('node:http')
+const { once } = require('node:events')
+const { Agent, Pool, fetch } = require('../..')
+
+// Regression for #5600 / #5606:
+// Reusing an idle keep-alive socket must not stall behind the poll phase.
+// Progress is asserted via logical I/O events (server 'connection' / 'request'),
+// not wall-clock thresholds. If idle-socket validation is deferred with an
+// unref'd setImmediate, the poll phase blocks on the re-ref'd socket and this
+// test hits the timeout instead of completing.
+//
+// The buggy setImmediate path stalls ~TICK_MS (~499ms) per reuse when the
+// event loop is idle (woken only by undici's fast-timer tick). Five reuses
+// therefore take well over 1s when regressed, and a few dozen ms when fixed.
+
+const REUSES = 5
+
+test('reusing an idle keep-alive socket must not stall', { timeout: 1000 }, async (t) => {
+  let connections = 0
+
+  const server = createServer((req, res) => {
+    res.writeHead(200, { 'content-length': 2 })
+    res.end('ok')
+  })
+
+  server.on('connection', () => {
+    connections++
+  })
+
+  server.listen(0)
+  await once(server, 'listening')
+
+  const pool = new Pool(`http://127.0.0.1:${server.address().port}`, {
+    connections: 1
+  })
+
+  t.after(async () => {
+    await pool.close()
+    server.close()
+  })
+
+  // Establish the keep-alive connection.
+  {
+    const res = await pool.request({ path: '/0', method: 'GET' })
+    assert.strictEqual(await res.body.text(), 'ok')
+  }
+  assert.strictEqual(connections, 1)
+
+  // Each reuse is gated on the server observing the request. Between
+  // iterations the client socket is idle/unref'd, which is the state that
+  // triggers idle-socket validation on the next dispatch.
+  for (let i = 1; i <= REUSES; i++) {
+    const requested = once(server, 'request')
+    const resPromise = pool.request({ path: `/${i}`, method: 'GET' })
+    // Suppress unhandled rejection if the test times out mid-request.
+    resPromise.catch(() => {})
+
+    await requested
+
+    const res = await resPromise
+    assert.strictEqual(await res.body.text(), 'ok')
+    assert.strictEqual(connections, 1, 'keep-alive socket must be reused')
+  }
+})
+
+test('fetch reusing an idle keep-alive socket must not stall', { timeout: 1000 }, async (t) => {
+  let connections = 0
+
+  const server = createServer((req, res) => {
+    res.writeHead(200, { 'content-length': 2 })
+    res.end('ok')
+  })
+
+  server.on('connection', () => {
+    connections++
+  })
+
+  server.listen(0)
+  await once(server, 'listening')
+
+  const url = `http://127.0.0.1:${server.address().port}`
+  const agent = new Agent({
+    connections: 1,
+    pipelining: 1,
+    keepAliveTimeout: 60_000
+  })
+
+  t.after(async () => {
+    await agent.close()
+    server.close()
+  })
+
+  {
+    const res = await fetch(`${url}/0`, { dispatcher: agent })
+    assert.strictEqual(await res.text(), 'ok')
+  }
+  assert.strictEqual(connections, 1)
+
+  for (let i = 1; i <= REUSES; i++) {
+    const requested = once(server, 'request')
+    const resPromise = fetch(`${url}/${i}`, { dispatcher: agent })
+    resPromise.catch(() => {})
+
+    await requested
+
+    const res = await resPromise
+    assert.strictEqual(await res.text(), 'ok')
+    assert.strictEqual(connections, 1, 'keep-alive socket must be reused')
+  }
+})