fix(h2): allow stream-bodied requests to multiplex on a busy session

lib/dispatcher/client-h2.js

@@ -322,16 +322,6 @@ function connectH2 (client, socket) {
           // Don't dispatch an upgrade until all preceding requests have completed.
           // Possibly, we do not have remote settings confirmed yet.
           if ((request.upgrade === 'websocket' || request.method === 'CONNECT') && session[kRemoteSettings] === false) return true
-          // Request with stream or iterator body can error while other requests
-          // are inflight and indirectly error those as well.
-          // Ensure this doesn't happen by waiting for inflight
-          // to complete before dispatching.
-
-          // Request with stream or iterator body cannot be retried.
-          // Ensure that no other requests are inflight and
-          // could cause failure.
-          if (util.bodyLength(request.body) !== 0 &&
-            (util.isStream(request.body) || util.isAsyncIterable(request.body) || util.isFormDataLike(request.body))) return true
         } else {
           return (request.upgrade === 'websocket' || request.method === 'CONNECT') && session[kRemoteSettings] === false
         }

test/http2-pipelining-default.js

@@ -1,12 +1,12 @@
 'use strict'
 
 const { test, after } = require('node:test')
-const { createSecureServer } = require('node:http2')
+const { createSecureServer, createServer } = require('node:http2')
 const { once } = require('node:events')
 const { tspl } = require('@matteo.collina/tspl')
 const pem = require('@metcoder95/https-pem')
 
-const { Client, Pool } = require('..')
+const { Agent, Client, Pool, fetch } = require('..')
 
 test('h2 client multiplexes concurrent requests by default (#4143)', async t => {
   const N = 5
@@ -113,6 +113,137 @@
   await t.completed
 })
 
+test('fetch POST multiplexes while an SSE stream is open on the same h2 session (#5524)', async t => {
+  t = tspl(t, { plan: 4 })
+
+  const server = createServer()
+  const paths = []
+  const sessions = new Set()
+  let eventsOpened
+  const eventsOpenedPromise = new Promise(resolve => {
+    eventsOpened = resolve
+  })
+
+  server.on('session', session => {
+    sessions.add(session)
+  })
+
+  server.on('stream', (stream, headers) => {
+    paths.push(headers[':path'])
+
+    if (headers[':path'] === '/events') {
+      stream.respond({ ':status': 200, 'content-type': 'text/event-stream' })
+      stream.write(': ping\n\n')
+      eventsOpened()
+      return
+    }
+
+    stream.respond({ ':status': 200, 'content-type': 'application/json' })
+    stream.end('{"ok":true}')
+  })
+
+  await once(server.listen(0, '127.0.0.1'), 'listening')
+  after(() => server.close())
+
+  const dispatcher = new Agent({ useH2c: true })
+  const sse = new AbortController()
+  after(async () => {
+    sse.abort()
+    await dispatcher.close()
+  })
+
+  const origin = `http://127.0.0.1:${server.address().port}`
+
+  const warmup = await fetch(`${origin}/warmup`, {
+    method: 'POST',
+    body: '{"warmup":true}',
+    dispatcher
+  })
+  await warmup.text()
+
+  fetch(`${origin}/events`, {
+    dispatcher,
+    signal: sse.signal
+  }).catch(() => {})
+  await eventsOpenedPromise
+
+  const response = await fetch(`${origin}/rpc`, {
+    method: 'POST',
+    body: '{"ok":true}',
+    dispatcher,
+    signal: AbortSignal.timeout(5000)
+  })
+
+  t.strictEqual(response.status, 200)
+  t.strictEqual(await response.text(), '{"ok":true}')
+  t.deepStrictEqual(paths, ['/warmup', '/events', '/rpc'])
+  t.strictEqual(sessions.size, 1)
+
+  await t.completed
+})
+
+test('fetch POST bodies dispatch concurrently on the same h2 session instead of serializing (#5494)', async t => {
+  t = tspl(t, { plan: 3 })
+
+  const server = createServer()
+  const sessions = new Set()
+  const streams = []
+  let resolveSecondArrived
+  const secondArrived = new Promise(resolve => {
+    resolveSecondArrived = resolve
+  })
+
+  server.on('session', session => {
+    sessions.add(session)
+  })
+
+  server.on('stream', stream => {
+    // Hold every stream open instead of responding immediately, so a
+    // serialized client would never let the second request's headers
+    // reach the server until the first one's stream is released.
+    streams.push(stream)
+    if (streams.length === 2) {
+      resolveSecondArrived()
+    }
+  })
+
+  await once(server.listen(0, '127.0.0.1'), 'listening')
+  after(() => server.close())
+
+  const dispatcher = new Agent({ useH2c: true })
+  after(async () => {
+    await dispatcher.close()
+  })
+
+  const origin = `http://127.0.0.1:${server.address().port}`
+
+  const first = fetch(origin, { method: 'POST', body: '{"first":1}', dispatcher })
+  // Wait for the first request's stream to actually open before dispatching
+  // the second -- this reproduces the exact ordering #5494 reported
+  // (one bodied request already in flight when the next one is dispatched).
+  while (streams.length < 1) {
+    await new Promise(resolve => setImmediate(resolve))
+  }
+
+  const second = fetch(origin, { method: 'POST', body: '{"second":1}', dispatcher, signal: AbortSignal.timeout(5000) })
+
+  // If the second request is stuck behind the first (the bug), its stream
+  // never opens and this hangs until the timeout signal aborts it.
+  await secondArrived
+
+  for (const stream of streams) {
+    stream.respond({ ':status': 200 })
+    stream.end('{"ok":true}')
+  }
+
+  const [firstRes, secondRes] = await Promise.all([first, second])
+  t.strictEqual(firstRes.status, 200)
+  t.strictEqual(secondRes.status, 200)
+  t.strictEqual(sessions.size, 1)
+
+  await t.completed
+})
+
 test('Client#pipelining keeps its h1 (RFC7230) semantic on an h2 client', async t => {
   t = tspl(t, { plan: 2 })