Retry fetch when an idle H2 session socket closes

lib/web/fetch/index.js

@@ -67,6 +67,12 @@ const { bytesMatch } = require('../subresource-integrity/subresource-integrity')
 const { isomorphicEncode } = require('../infra')
 
 const GET_OR_HEAD = ['GET', 'HEAD']
+const retriableFetchNetworkErrorCodes = new Set([
+  'ECONNRESET',
+  'EPIPE',
+  'UND_ERR_SOCKET'
+])
+const kFetchNetworkErrorRetry = Symbol('fetch network error retry')
 
 const defaultUserAgent = typeof __UNDICI_IS_NODE__ !== 'undefined' || typeof esbuildDetection !== 'undefined'
   ? 'node'
@@ -151,6 +157,34 @@ class Fetch extends EE {
   }
 }
 
+function createFetchConnection (fetchParams) {
+  fetchParams.controller.connection = {
+    abort: null,
+    destroyed: false,
+    destroy (err, abort = true) {
+      if (!this.destroyed) {
+        this.destroyed = true
+        if (abort) {
+          this.abort?.(err ?? new DOMException('The operation was aborted.', 'AbortError'))
+        }
+      }
+    }
+  }
+}
+
+function canReplayRequestBody (request) {
+  return request.body == null || request.body.source != null
+}
+
+function isRetriableFetchNetworkError (error) {
+  return retriableFetchNetworkErrorCodes.has(error?.code) ||
+    retriableFetchNetworkErrorCodes.has(error?.cause?.code)
+}
+
+function waitForFetchRetry () {
+  return new Promise(resolve => setTimeout(resolve, 0))
+}
+
 function handleFetchDone (response) {
   finalizeAndReportTiming(response, 'fetch')
 }
@@ -1815,18 +1849,7 @@ async function httpNetworkFetch (
 ) {
   assert(!fetchParams.controller.connection || fetchParams.controller.connection.destroyed)
 
-  fetchParams.controller.connection = {
-    abort: null,
-    destroyed: false,
-    destroy (err, abort = true) {
-      if (!this.destroyed) {
-        this.destroyed = true
-        if (abort) {
-          this.abort?.(err ?? new DOMException('The operation was aborted.', 'AbortError'))
-        }
-      }
-    }
-  }
+  createFetchConnection(fetchParams)
 
   // 1. Let request be fetchParams’s request.
   const request = fetchParams.request
@@ -2000,6 +2023,22 @@ async function httpNetworkFetch (
       response = makeResponse({ status, statusText, headersList })
     }
   } catch (err) {
+    if (err?.[kFetchNetworkErrorRetry]) {
+      fetchParams.controller.connection.destroy(err, false)
+
+      await waitForFetchRetry()
+
+      if (isCancelled(fetchParams)) {
+        return makeAppropriateNetworkError(fetchParams, err)
+      }
+
+      if (request.body != null) {
+        request.body = safelyExtractBody(request.body.source)[0]
+      }
+
+      return httpNetworkFetch(fetchParams, includeCredentials, true)
+    }
+
     // 10. If aborted, then:
     if (err.name === 'AbortError') {
       // 1. If connection uses HTTP/2, then transmit an RST_STREAM frame.
@@ -2376,6 +2415,18 @@ async function httpNetworkFetch (
               return
             }
 
+            if (
+              this.body == null &&
+              !forceNewConnection &&
+              request.mode !== 'websocket' &&
+              canReplayRequestBody(request) &&
+              isRetriableFetchNetworkError(error)
+            ) {
+              error[kFetchNetworkErrorRetry] = true
+              reject(error)
+              return
+            }
+
             this.body?.destroy(error)
 
             fetchParams.controller.terminate(error)

test/fetch/http2.js

@@ -9,7 +9,7 @@ const { Readable } = require('node:stream')
 const { test } = require('node:test')
 const pem = require('@metcoder95/https-pem')
 
-const { Client, fetch, Headers } = require('../..')
+const { Agent, Client, fetch, Headers } = require('../..')
 
 const { closeClientAndServerAsPromise } = require('../utils/node-http')
 
@@ -123,6 +123,68 @@ test('[Fetch] Simple GET with h2', async (t) => {
   t.assert.strictEqual(response.statusText, '')
 })
 
+test('[Fetch] retries h2 fetch after idle session socket closes', async (t) => {
+  const server = createSecureServer(await pem.generate({ opts: { keySize: 2048 } }))
+  const requestBodies = []
+  const sockets = []
+  let streams = 0
+
+  server.on('secureConnection', (socket) => {
+    socket.on('error', () => {})
+    sockets.push(socket)
+  })
+  server.on('sessionError', () => {})
+  server.on('tlsClientError', () => {})
+  server.on('stream', async (stream) => {
+    streams++
+    const chunks = []
+
+    for await (const chunk of stream) {
+      chunks.push(chunk)
+    }
+
+    requestBodies.push(Buffer.concat(chunks).toString())
+    stream.respond({ ':status': 200 })
+    stream.end('ok')
+  })
+
+  t.plan(4)
+
+  server.listen(0)
+  await once(server, 'listening')
+
+  const dispatcher = new Agent({
+    connections: 1,
+    connect: {
+      rejectUnauthorized: false
+    },
+    allowH2: true
+  })
+
+  t.after(async () => {
+    for (const socket of sockets) {
+      socket.destroy()
+    }
+
+    await dispatcher.close()
+    server.close()
+  })
+
+  const first = await fetch(`https://localhost:${server.address().port}/`, { dispatcher })
+  t.assert.strictEqual(await first.text(), 'ok')
+
+  sockets[0].destroy()
+
+  const second = await fetch(`https://localhost:${server.address().port}/`, {
+    dispatcher,
+    method: 'POST',
+    body: 'replayed body'
+  })
+  t.assert.strictEqual(await second.text(), 'ok')
+  t.assert.strictEqual(streams, 2)
+  t.assert.strictEqual(requestBodies[1], 'replayed body')
+})
+
 test('[Fetch] Should handle h2 request with body (string or buffer)', async (t) => {
   const server = createSecureServer(await pem.generate({ opts: { keySize: 2048 } }))
   const expectedBody = 'hello from client!'